88 lines
2.1 KiB
Python
88 lines
2.1 KiB
Python
import os
|
|
import lark_oapi as lark
|
|
|
|
import lark_oapi as lark
|
|
|
|
from flask import Flask
|
|
from dotenv import load_dotenv
|
|
from lark_oapi.adapter.flask import *
|
|
from lark_oapi.api.im.v1 import *
|
|
from lark_oapi.api.bitable.v1 import *
|
|
|
|
|
|
load_dotenv()
|
|
|
|
app_id = os.environ["APP_ID"]
|
|
app_secret = os.environ["APP_SECRET"]
|
|
|
|
encrypt_key = os.environ["ENCRYPT_KEY"]
|
|
verification_token = os.environ["VERIFICATION_TOKEN"]
|
|
|
|
bitable_token = os.environ["BITABLE_TOKEN"]
|
|
table_id = os.environ["TABLE_ID"]
|
|
|
|
# 创建client
|
|
client = lark.Client.builder().app_id(app_id).app_secret(app_secret).build()
|
|
|
|
|
|
app = Flask(__name__)
|
|
|
|
|
|
def do_p2_im_message_receive_v1(data: P2ImMessageReceiveV1) -> None:
|
|
print(lark.JSON.marshal(data))
|
|
record = {}
|
|
record["问题概述"] = data.event.message.content
|
|
record["创建人"] = data.event.sender.sender_id.open_id
|
|
add_record(record)
|
|
|
|
|
|
def do_customized_event(data: lark.CustomizedEvent) -> None:
|
|
print(lark.JSON.marshal(data))
|
|
|
|
|
|
handler = (
|
|
lark.EventDispatcherHandler.builder(
|
|
encrypt_key, verification_token, lark.LogLevel.DEBUG
|
|
)
|
|
.register_p2_im_message_receive_v1(do_p2_im_message_receive_v1)
|
|
.register_p1_customized_event("message", do_customized_event)
|
|
.build()
|
|
)
|
|
|
|
|
|
def add_record(data):
|
|
# 构造请求对象
|
|
request: CreateAppTableRecordRequest = (
|
|
CreateAppTableRecordRequest.builder()
|
|
.app_token(bitable_token)
|
|
.table_id(table_id)
|
|
.request_body(AppTableRecord.builder().fields(data).build())
|
|
.build()
|
|
)
|
|
|
|
# 发起请求
|
|
response: CreateAppTableRecordResponse = client.bitable.v1.app_table_record.create(
|
|
request
|
|
)
|
|
|
|
# 处理失败返回
|
|
if not response.success():
|
|
lark.logger.error(
|
|
f"client.bitable.v1.app_table_record.create failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}"
|
|
)
|
|
return
|
|
|
|
# 处理业务结果
|
|
lark.logger.info(lark.JSON.marshal(response.data, indent=4))
|
|
|
|
|
|
@app.route("/event", methods=["POST"])
|
|
def event():
|
|
resp = handler.do(parse_req())
|
|
|
|
return parse_resp(resp)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(port=8080, debug=True, host="0.0.0.0")
|