BXY0414/gmv-analytics
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

通过应用机器人推送到飞书群中

Browse Source
This commit is contained in:
BXY0414
2026-03-13 18:19:17 +08:00
parent 679edd1f58
commit fff6504412
7 changed files with 276 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
project_bitable_common.py
+187 -5
View File
@@ -17,9 +17,8 @@ from openpyxl import load_workbook
BASE_DIR = Path(__file__).resolve().parent
ENV_FILE = BASE_DIR / ".env"
DEFAULT_MAPPING_FILE = BASE_DIR / "店铺映射表.xlsx"
DEFAULT_START_DATE = "2026-03-01"
DEFAULT_END_DATE = "2026-03-01"
DEFAULT_STORE_FIELD_NAME = "业务店铺名称"
DEFAULT_START_DATE = "2026-02-25"
DEFAULT_STORE_FIELD_NAME = "店铺名称"
STORE_FIELD_CANDIDATES = ["业务店铺名称", "店铺名称", "门店名称", "店铺", "名称"]
SUPPORTED_PROJECTS = ["大眼睛", "羽心堂", "MMMJ", "EM", "春回盏", "BROKEN ZOO"]
@@ -278,6 +277,42 @@ def get_mapping_file() -> Path:
return Path(os.getenv("PROJECT_MAPPING_FILE", str(DEFAULT_MAPPING_FILE)))
def get_store_field_name() -> str:
return os.getenv("FEISHU_STORE_FIELD_NAME", DEFAULT_STORE_FIELD_NAME)
def get_default_start_date() -> str:
return os.getenv("SYNC_START_DATE", DEFAULT_START_DATE)
def get_default_end_date() -> str:
return (date.today() - timedelta(days=1)).isoformat()
def get_group_webhook_url() -> str:
candidates = [
"FEISHU_BOT_WEBHOOK",
"FEISHU_ROBOT_WEBHOOK",
"FEISHU_WEBHOOK",
"LARK_BOT_WEBHOOK",
"LARK_ROBOT_WEBHOOK",
"GROUP_WEBHOOK",
]
for key in candidates:
value = os.getenv(key)
if value:
return value
return ""
def get_feishu_chat_id() -> str:
return os.getenv("FEISHU_CHAT_ID", "")
def get_feishu_report_image_path() -> str:
return os.getenv("FEISHU_REPORT_IMAGE_PATH", "")
def resolve_store_field_name(fields: list[dict[str, Any]], preferred_field_name: str) -> str:
field_names = [str(item.get("field_name") or "") for item in fields]
if preferred_field_name in field_names:
@@ -330,7 +365,15 @@ def load_store_mapping(path: Path) -> dict[str, str]:
wb = load_workbook(path, data_only=True)
ws = wb.active
mapping: dict[str, str] = {}
for business_name, bi_name in ws.iter_rows(min_row=2, values_only=True):
for row in ws.iter_rows(min_row=2, values_only=True):
if not row:
continue
if len(row) >= 3:
_, business_name, bi_name = row[:3]
elif len(row) >= 2:
business_name, bi_name = row[:2]
else:
continue
if not business_name or not bi_name:
continue
mapping[str(bi_name).strip()] = str(business_name).strip()
@@ -610,6 +653,146 @@ def load_table_name_overrides(path_text: str) -> dict[str, str]:
return {str(key): str(value) for key, value in payload.items()}
def send_feishu_sync_card(
webhook_url: str,
title: str,
base_url: str,
start_date: date,
end_date: date,
) -> None:
if not webhook_url:
log_step("未配置飞书群机器人 webhook，跳过群消息推送")
return
card = {
"msg_type": "interactive",
"card": {
"config": {"wide_screen_mode": True},
"header": {
"template": "blue",
"title": {"tag": "plain_text", "content": title},
},
"elements": [
{
"tag": "div",
"text": {
"tag": "lark_md",
"content": (
f"**同步范围**\n{start_date.isoformat()} 至 {end_date.isoformat()}\n\n"
f"**多维表格**\n[点击打开多维表格]({base_url})"
),
},
}
],
},
}
log_step("开始推送飞书群卡片消息")
resp = requests.post(webhook_url, json=card, timeout=30)
resp.raise_for_status()
payload = resp.json()
if payload.get("code") not in (0, None):
raise RuntimeError(f"飞书群消息推送失败: {payload}")
log_step("飞书群卡片消息推送完成")
def upload_feishu_image(image_path: str) -> str:
if not image_path:
return ""
path = Path(image_path)
if not path.exists():
raise FileNotFoundError(f"未找到预览图文件: {path}")
token = get_feishu_token()
headers = {"Authorization": f"Bearer {token}"}
with path.open("rb") as image_file:
files = {"image": (path.name, image_file, "application/octet-stream")}
data = {"image_type": "message"}
log_step(f"开始上传飞书消息图片: {path}")
resp = requests.post("https://open.feishu.cn/open-apis/im/v1/images", headers=headers, data=data, files=files, timeout=60)
resp.raise_for_status()
payload = resp.json()
if payload.get("code") != 0:
raise RuntimeError(f"飞书图片上传失败: {payload}")
image_key = ((payload.get("data") or {}).get("image_key")) or ""
if not image_key:
raise RuntimeError(f"飞书图片上传成功但未返回 image_key: {payload}")
log_step(f"飞书消息图片上传完成: image_key={image_key}")
return image_key
def send_feishu_app_card(
chat_id: str,
title: str,
base_url: str,
start_date: date,
end_date: date,
image_path: str = "",
) -> None:
if not chat_id:
log_step("未配置 FEISHU_CHAT_ID，跳过应用机器人卡片推送")
return
token = get_feishu_token()
image_key = upload_feishu_image(image_path) if image_path else ""
elements: list[dict[str, Any]] = [
{
"tag": "div",
"text": {
"tag": "lark_md",
"content": (
f"**同步范围**\n{start_date.isoformat()} 至 {end_date.isoformat()}\n\n"
f"**多维表格**\n[{title}]({base_url})"
),
},
}
]
if image_key:
elements.append({"tag": "img", "img_key": image_key, "alt": {"tag": "plain_text", "content": "报表预览图"}})
elements.append(
{
"tag": "action",
"actions": [
{
"tag": "button",
"text": {"tag": "plain_text", "content": "打开仪表盘"},
"type": "primary",
"url": base_url,
}
],
}
)
message = {
"receive_id": chat_id,
"msg_type": "interactive",
"content": json.dumps(
{
"config": {"wide_screen_mode": True},
"header": {"template": "blue", "title": {"tag": "plain_text", "content": title}},
"elements": elements,
},
ensure_ascii=False,
),
}
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json; charset=utf-8",
}
log_step(f"开始通过应用机器人发送群卡片: chat_id={chat_id}")
resp = requests.post(
"https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id",
headers=headers,
json=message,
timeout=60,
)
try:
payload = resp.json()
except ValueError:
resp.raise_for_status()
raise RuntimeError(f"应用机器人发送群卡片失败，返回非 JSON: status={resp.status_code}, body={resp.text[:500]}")
if resp.status_code >= 400:
raise RuntimeError(f"应用机器人发送群卡片失败: status={resp.status_code}, payload={payload}")
if payload.get("code") != 0:
raise RuntimeError(f"应用机器人发送群卡片失败: {payload}")
log_step("应用机器人群卡片发送完成")
def find_target_table(
tables: list[TableInfo],
project: str,
@@ -916,4 +1099,3 @@ def run_gmv_sync(
preferred_store_field_name=store_field_name,
dry_run=dry_run,
)