634 lines
21 KiB
Python
634 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
import json
|
|
import os
|
|
import ssl
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Callable, 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
|
|
|
|
|
|
def create_feishu_tls12_context() -> ssl.SSLContext:
|
|
"""Create a verified TLS 1.2 context for Feishu API requests."""
|
|
|
|
context = ssl.create_default_context()
|
|
context.minimum_version = ssl.TLSVersion.TLSv1_2
|
|
context.maximum_version = ssl.TLSVersion.TLSv1_2
|
|
return context
|
|
|
|
|
|
FEISHU_HTTP_OPENER = request.build_opener(request.HTTPSHandler(context=create_feishu_tls12_context()))
|
|
|
|
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,
|
|
}
|
|
|
|
LOGGER_WRITER: Optional[Callable[[str, str], None]] = None
|
|
|
|
|
|
def log(message: str, *, level: str = "INFO") -> None:
|
|
if LOGGER_WRITER is not None:
|
|
LOGGER_WRITER(level, message)
|
|
return
|
|
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
print(f"[{timestamp}] [{level}] {message}")
|
|
|
|
|
|
def set_logger_writer(writer: Optional[Callable[[str, str], None]]) -> None:
|
|
global LOGGER_WRITER
|
|
LOGGER_WRITER = writer
|
|
|
|
|
|
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 FEISHU_HTTP_OPENER.open(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, ssl.SSLError) 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())
|