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: lark.logger.info(f"receive message: {lark.JSON.marshal(data)}") # print(client, bitable_token, table_id, cache_dir, data) message_type = data.event.message.message_type content = json.loads(data.event.message.content) sender_id = data.event.sender.sender_id.open_id message_id = data.event.message.message_id thread_id = data.event.message.thread_id clear_data = {} clear_data["message_type"] = message_type clear_data["content"] = content clear_data["sender_id"] = sender_id clear_data["message_id"] = message_id clear_data["thread_id"] = thread_id 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, clear_data ) print(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] target_actions = ["record_edited", "record_added"] actions = [ action for action in action_list if action.get("action") == "record_edited" ] if not (set(target_actions) <= set(actions_name)): return data if table_id != "tblESHsFZwtiFROB": return data if not actions: return data 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") print(f"table_id: {table_id}") print(f"record_id: {record_id}") print(f"target_action: {target_action.get('action')}") print(action_data) table_id = "tblESHsFZwtiFROB" process_new_demand.delay( app_id, app_secret, table_id, record_id, action_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_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), }, } app.extensions["celery"] = celery_app return celery_app