作者：张浩
推荐理由
通过飞书开放能力，实现微信视频号小店虚拟商品自动发货。
一、关于我
- 我是磁场 App 的后端开发，工作中主要语言是 Ruby，在飞书上会通过 Python 和 Go 做集成开发，帮助企业提升效率。
二、需求分析
- 磁场 App 是一个线上学习社区，上面有我们的自营社区，自营社区通过微信视频号直播带货，销售自营社区的兑换码，用户在购买虚拟商品（即兑换码）后，需要及时给用户发送兑换码短信。
- 用户在视频号小店下单后，运营往往要手动一个一个地给用户发送，这样效率低且体验不好，需求是要将用户在视频号下单后发货的流程全部自动化，来提高运营效率和体验。
三、方案调研
- 方案一：使用视频号小店服务市场发货服务
- 视频号小店很快上线了服务市场，有提供虚拟商品自动发货服务，使用起来非常方便，但需要运营在磁场运营后台批量生成兑换码，再粘贴到服务商后台，实际使用起来比较繁琐，关键问题是用户数据无法与兑换码一一匹配，方案经过测试不满足需求，不予采纳。
- 方案二：自研发货系统
- 考虑自研发货系统，中小企业研发资源有限，要把资源用在最重要的事情上，一方面自研成本相对过高，另一方面考虑到开发后复用率很低，综合考虑不自研发货系统。
- 方案三：借助飞书的能力实现自动发货
- 飞书多维表格相当于小型应用，借助飞书多维表格和开放平台能力，再结合我们磁场 App，经过调研和测试，可以实现视频号小店自动发货功能，决定采用该方案。
四、开发流程
- 步骤一：创建飞书自建应用
- 参考创建企业自建应用开发文档，在飞书开放平台创建自建应用，名字为【视频号自动发货】，需要在【步骤二】添加到多维表格中，获取执行多维表格的各种权限。
- 关键点：
- 注意获取飞书多维表格相关权限
- 步骤二：创建飞书多维表格
- 创建两个飞书多维表格，一个是【视频号自动发货（正式）】，另一个是【视频号自动发货记录（正式）】，前者添加【视频号自动发货】应用，用于对外发送 HTTPS 请求，向磁场服务端发送执行发送短信的请求；后者做自动发送短信的记录，方便运营查询短信发送状态、兑换码使用情况、兑换码信息等内容。
- 关键点
- 【自动化】中配置发送 HTTPS
- 【文档应用】中注意要添加【视频号小店自动发货】自建应用和【多维表格助手】，来获取对多维表格的访问权限
- 步骤三：在Replit创建Repls
- 关键点
- 循环执行逻辑 run_schedule_with_loop.py 文件
import time
import schedule
import wechat_channel_auto_delivery
def main_scheduler():
while True:
wechat_channel_auto_delivery.time_job()
# 每半小时执行一次
time.sleep(1800)
if __name__ == '__main__':
print("执行main_scheduler")
main_scheduler()
- 自动发货业务逻辑 wechat_channel_auto_delivery.py 文件
import time
import items
import oapi
def handler():
access_token = oapi.get_wechat_access_token(items.WC_NULISHEHUI_APPID, items.WC_NULISHEHUI_APPSECRET)
if not access_token:
print('Failed to get access token')
return
now_date = int(time.time())
now_time = time.localtime(now_date)
now_strf_time = time.strftime("%Y-%m-%d %H:%M:%S", now_time)
start_time = int(now_date - 24 * 3600 - 5)
end_time = int(now_date)
print(f"\nINFO {now_strf_time} | begin to run.")
order_list = oapi.get_wechat_channel_order_list(access_token, start_time, end_time)
if not order_list:
return
for order in order_list:
order_info = oapi.get_wechat_channel_order_details(access_token, order)
shoppingname = order_info['order_detail']['product_infos'][0]['title'] #商品名称
phone = order_info['order_detail']['delivery_info']['address_info']['virtual_order_tel_number'] # 手机号码
pay_time = order_info['order_detail']['pay_info']['pay_time']*1000 #支付时间
records_resp = {
"records":[{"fields":{"订单":order, "phone":phone, "商品名称":shoppingname, "pay_time":pay_time}}]}
deliver_method = order_info["order_detail"]["delivery_info"]["deliver_method"]
if deliver_method == 1:
oapi.batch_create_feishu_bitable_records(records_resp, items.WC_FEISHU_BITABLE_APP_TOKEN, items.WC_FEISHU_BITABLE_PERSONAL_TOKEN, items.WC_FEISHU_BITABLE_TABLE_ID)
sku_id = order_info['order_detail']['product_infos'][0]['sku_id']
product_id = order_info['order_detail']['product_infos'][0]['product_id']
oapi.wechat_channel_order_deliver(access_token, order, sku_id, product_id)
oapi.send_feishu_group_message(order,phone,shoppingname,pay_time)
def time_job():
handler()
- 定义的 API 接口 oapi.py 文件
from baseopensdk import BaseClient
from baseopensdk.api.base.v1 import BatchUpdateAppTableRecordRequest, ListAppTableRecordRequest, \
ListAppTableFieldRequest, BatchCreateAppTableRecordRequest
import items
import requests
import json
import time
# 获取 client
def client(app_token, personal_token):
result = BaseClient.builder().app_token(app_token).personal_base_token(personal_token).build()
return result
# 批量创建多维表格记录
def batch_create_feishu_bitable_records(create_records, app_token, personal_token, table_id):
request = BatchCreateAppTableRecordRequest.builder().table_id(table_id).request_body(create_records).build()
response = client(app_token, personal_token).base.v1.app_table_record.batch_create(request)
if not response.success():
return False
return response.data
# 获取微信access_token
def get_wechat_access_token(appid, appsecret):
url = 'https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid=' + appid + '&secret=' + appsecret
response = requests.get(url)
data = json.loads(response.text)
access_token = data.get('access_token')
return access_token
# 获取视频号小店订单列表
def get_wechat_channel_order_list(access_token, start_time, end_time):
headers = { 'Content-Type': 'application/json' }
data = {
"create_time_range": { "start_time": start_time, "end_time": end_time },
"status": 20,
"page_size": 99,
}
response = requests.post(url, headers=headers, data=json.dumps(data))
if response.status_code == 200:
result = json.loads(response.text)
if result.get('errcode') == 0:
order_list = result.get('order_id_list')
return order_list
return None
# 获取订单详情
def get_wechat_channel_order_details(access_token, order_id):
headers = { 'Content-Type': 'application/json' }
data = { "order_id": order_id }
response = requests.post(url, headers=headers, data=json.dumps(data))
if response.status_code == 200:
result = json.loads(response.text)
if result.get('errcode') == 0:
order_details = result.get('order')
return order_details
return None
# 修改视频号小店订单发货状态
def wechat_channel_order_deliver(access_token, order, sku_id, product_id):
access_token, order_id, sku_id, product_id = access_token, int(order), int(sku_id), int(product_id)
headers = {'Content-Type': 'application/json'}
data = {
"order_id": order_id,
"delivery_list": [{
"deliver_type": 3,
"product_infos": [{
"product_cnt": 1,
"product_id": product_id,
"sku_id": sku_id
}]}]}
try:
requests.post(url, headers=headers, data=json.dumps(data))
print("Deliver order success")
except Exception as e:
print("Deliver failed:", e)
# 发送飞书群组消息
def send_feishu_group_message(order, phone, shoppingname, pay_time):
pay_time1 = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(pay_time/1000))
headers = {'Content-Type': 'application/json'}
data = {
"msg_type": "interactive",
"card": {
"elements": [{
"tag": "div",
"text": {
"content": f"商品:'{shoppingname}'\n 订单号：'{order}'\n 手机号：'{phone}'\n 支付时间：'{pay_time1}'",
"tag": "lark_md"
}
}],
"header": {
"title": {
"content": "视频号小店发货通知",
"tag": "plain_text"
}
}
}
}
try:
requests.post(url, headers=headers, data=json.dumps(data))
print("feishu msg push")
except Exception as e:
print("Failed to send:", e)
- 定义变量的 items.py文件
# 飞书【视频号发货通知】群
WC_FEISHU_GROUP_ID=""
# 视频号APPID和APPSECRET
WC_NULISHEHUI_APPID = ""
WC_NULISHEHUI_APPSECRET = ""
# 微信视频号订单发货信息-飞书多维表格授权码
WC_FEISHU_BITABLE_PERSONAL_TOKEN = ""
WC_FEISHU_BITABLE_APP_TOKEN = ""
WC_FEISHU_BITABLE_TABLE_ID = ""
WC_FEISHU_BITABLE_VIEW_ID = ""
PHONE = "phone"
WECHAT_CHANNEL_OEDER = "订单"
注：磁场服务端开发语言为 Ruby，步骤四和步骤五为 Ruby 代码示例。
- 步骤四：磁场集成发送短信服务（选看）
- 磁场主要集成短信发送服务，目前使用 Submail 处理发送短信业务逻辑，再新建一个飞书群组并创建机器人，兑换码被使用后发送一条消息，同步信息形成闭环。
- 详细请看🫱：使用飞书实现微信视频号虚拟商品自动发货（步骤四和步骤五）
- 步骤五：从 Replit 迁移到 磁场Kubernetes集群（选看）
- 自动发货服务在 Replit 运行一个月后，在生产业务中验证了可行性，打算将服务迁移到磁场的 Kubernetes 集群中，使用 Gitlab 社区开源版，以下是部署过程的主要配置文件，包括gitlab-ci.yml、Dockerfile和kubernetes/chart。
- 详细请看🫱：使用飞书实现微信视频号虚拟商品自动发货（步骤四和步骤五）
五、更多相关开发心得
- 多分享飞书最佳实践
- 基于飞书的开发是要帮助业务部门提升效率，业务同事那里可能有很多小的需求，但有时不知道飞书里都有什么，而开发那里有很多解决方案，两边需要通过某种方式对齐，分享飞书最佳实践到公司大群是很好的方式，让业务同事看到飞书能做哪些事情，挑自己想要的，有时可能不需要写代码。比如：飞书多维表格的公式字段类型，就很能解决将多个字段合并到一起的需求，但当时的业务同事并不知道。
- 多关注飞书产品动态
- 飞书迭代速度很快，如果需求暂时无法很好地解决，可以持续地关注飞书的更新，可能很快就会出方案了。比如：飞书多维表格出了【插件】功能后，就能更好地解决我们自动发货的需求。
- 多记录开发过程
- 每次基于飞书的集成开发，最好将开发过程记录到飞书文档中，方便过去的自己跟未来的自己协作，也可以分享出去，帮助其他人更好地使用飞书，就像这次写自动发货的文章，想让更多地人用好飞书提升效率，也能看到飞书还能这么用。
