227 lines
6.6 KiB
Python
227 lines
6.6 KiB
Python
import os
|
|
import json
|
|
import time
|
|
|
|
import lark_oapi as lark
|
|
from celery import Celery
|
|
from celery import Task
|
|
from celery.schedules import crontab
|
|
|
|
|
|
from flask import Flask
|
|
from dotenv import load_dotenv
|
|
from itertools import chain
|
|
from lark_oapi.adapter.flask import *
|
|
from lark_oapi.api.im.v1 import *
|
|
from lark_oapi.api.bitable.v1 import *
|
|
|
|
from app.task import *
|
|
from app.utils.utils import *
|
|
from app.task import *
|
|
from app.utils.fs_utils import AESCipher
|
|
|
|
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"]
|
|
|
|
# bitable_token = "IRggb4I0MaRffTsJ7qkcw1wxnBg"
|
|
# table_id = "tbl5J3sXBNIyNvJ8"
|
|
|
|
broker_url = os.environ["BROKER_URL"]
|
|
result_backend = os.environ["RESULT_BACKEND"]
|
|
|
|
|
|
# 创建client
|
|
client = lark.Client.builder().app_id(app_id).app_secret(app_secret).build()
|
|
|
|
current_dir = os.path.dirname(os.path.abspath(__file__))
|
|
cache_dir = os.path.join(current_dir, "cache")
|
|
|
|
if not os.path.exists(cache_dir):
|
|
os.makedirs(cache_dir)
|
|
|
|
|
|
def do_p2_im_message_receive_v1(data: P2ImMessageReceiveV1) -> None:
|
|
# print(client, bitable_token, table_id, cache_dir, data)
|
|
|
|
message_id = data.event.message.message_id
|
|
thread_id = data.event.message.thread_id
|
|
|
|
data = json.loads(lark.JSON.marshal(data))
|
|
print("=====================================")
|
|
print(f"thread_id: {thread_id}")
|
|
print(f"thread_id: {message_id}")
|
|
|
|
process_message_callback.delay(
|
|
app_id, app_secret, bitable_token, table_id, cache_dir, data
|
|
)
|
|
lark.logger.info(f"receive message: {lark.JSON.marshal(data)}")
|
|
print("=====================================")
|
|
|
|
|
|
def do_bitable_record_add(data) -> None:
|
|
|
|
event_data = lark.JSON.marshal(data)
|
|
print("=====================================")
|
|
print(event_data)
|
|
print("=====================================")
|
|
event = data.event
|
|
table_id = event.get("table_id")
|
|
action_list = event.get("action_list")
|
|
|
|
actions_name = [action.get("action") for action in action_list]
|
|
|
|
actions = [
|
|
action for action in action_list if action.get("action") == "record_edited"
|
|
]
|
|
print(actions)
|
|
|
|
# if not (set(target_actions) <= set(actions_name)):
|
|
# return data
|
|
|
|
if table_id != "tblESHsFZwtiFROB":
|
|
return data
|
|
|
|
if not actions:
|
|
return data
|
|
|
|
# 根据不同 "record_edited", "record_added" 的动作执行不同的任务
|
|
# 1. 如果包含 record_added 则创建需求文档
|
|
# 2. 如果包含 record_edited 并且判断 status 是否为已处理, 如果已处理则通过任务创建应用, 并在分类管理中添加一条记录, 在开发管理添加一条记录
|
|
|
|
target_action = actions[0]
|
|
record_id = target_action.get("record_id")
|
|
after_value: list = target_action.get("after_value")
|
|
|
|
action_data = {}
|
|
|
|
for field in after_value:
|
|
if field.get("field_id") == "fldIV2G9lv":
|
|
tmp = "".join(
|
|
[item["text"] for item in json.loads(field.get("field_value"))]
|
|
)
|
|
action_data["title"] = tmp
|
|
|
|
if field.get("field_id") == "fldk5ImODy":
|
|
tmp = "".join(
|
|
[item["text"] for item in json.loads(field.get("field_value"))]
|
|
)
|
|
action_data["background"] = tmp
|
|
|
|
if field.get("field_id") == "fldfjXnFVa":
|
|
tmp = "".join(
|
|
[item["text"] for item in json.loads(field.get("field_value"))]
|
|
)
|
|
action_data["features"] = tmp
|
|
|
|
if field.get("field_id") == "fld9QkmcS1":
|
|
action_data["category"] = field.get("field_value")
|
|
|
|
if field.get("field_id") == "fldopqyyCG":
|
|
action_data["status"] = field.get("field_value")
|
|
|
|
if field.get("field_id") == "fldaxMbywh":
|
|
# 有可能值为空
|
|
if field.get("field_value"):
|
|
action_data["deal_person"] = json.loads(field.get("field_value")).get("users")[0].get("name")
|
|
|
|
|
|
print(f"table_id: {table_id}")
|
|
print(f"record_id: {record_id}")
|
|
print(f"target_action: {target_action.get('action')}")
|
|
|
|
table_id = "tblESHsFZwtiFROB"
|
|
|
|
if "record_added" in actions_name:
|
|
process_new_demand.delay(app_id, app_secret, table_id, record_id, action_data)
|
|
return
|
|
|
|
if "record_edited" in actions_name and action_data["status"] == "optgKzdQp5":
|
|
# 映射对应的用户 id
|
|
user_name_id_map = {
|
|
"剑心": "80808032853078016",
|
|
"驿站": "59882836586950656",
|
|
"波音": "668720500384968704",
|
|
"虚竹": "620156920605310976",
|
|
}
|
|
user_id = user_name_id_map.get(action_data["deal_person"])
|
|
action_data["deal_person_id"] = user_id
|
|
process_new_robot.delay(app_id, app_secret, action_data)
|
|
return
|
|
|
|
|
|
handler = (
|
|
lark.EventDispatcherHandler.builder(
|
|
encrypt_key, verification_token, lark.LogLevel.DEBUG
|
|
)
|
|
.register_p2_im_message_receive_v1(do_p2_im_message_receive_v1)
|
|
.register_p2_customized_event(
|
|
"drive.file.bitable_record_changed_v1", do_bitable_record_add
|
|
)
|
|
.build()
|
|
)
|
|
|
|
|
|
def create_app() -> Flask:
|
|
app = Flask(__name__)
|
|
|
|
print("app is runging: ", time.time())
|
|
|
|
app.config.from_mapping(
|
|
CELERY=dict(
|
|
broker_url=broker_url,
|
|
result_backend=result_backend,
|
|
task_ignore_result=True,
|
|
broker_connection_retry_on_startup=True,
|
|
),
|
|
)
|
|
app.config.from_prefixed_env()
|
|
celery_init_app(app)
|
|
|
|
@app.route("/event", methods=["POST"])
|
|
def event():
|
|
print("receive event: ", time.time())
|
|
resp = handler.do(parse_req())
|
|
return parse_resp(resp)
|
|
|
|
@app.route("/index", methods=["GET"])
|
|
def index():
|
|
return {"test": "ok"}
|
|
|
|
return app
|
|
|
|
|
|
def celery_init_app(app: Flask) -> Celery:
|
|
class FlaskTask(Task):
|
|
def __call__(self, *args: object, **kwargs: object) -> object:
|
|
with app.app_context():
|
|
return self.run(*args, **kwargs)
|
|
|
|
celery_app = Celery(app.name, task_cls=FlaskTask)
|
|
celery_app.config_from_object(app.config["CELERY"])
|
|
celery_app.set_default()
|
|
celery_app.conf.timezone = "Asia/Shanghai"
|
|
celery_app.conf.beat_schedule = {
|
|
"sync-office-xbot-info": {
|
|
"task": "app.task.sync_xbot_activity_info",
|
|
"schedule": crontab(hour=19, minute=57),
|
|
},
|
|
}
|
|
celery_app.conf.beat_schedule = {
|
|
"check-activity-help-link": {
|
|
"task": "app.task.check_activity_help_link",
|
|
"schedule": crontab(hour=19, minute=55),
|
|
},
|
|
}
|
|
|
|
app.extensions["celery"] = celery_app
|
|
return celery_app
|