office_xbot_feishu/app/__init__.py

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]

    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")

        if field.get("field_id") == "fldopqyyCG":
            action_data["status"] = field.get("field_value")

        if field.get("field_id") == "fldaxMbywh":
            action_data["deal_person"] = (
                field.get("field_identity_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')}")
    print(action_data)

    table_id = "tblESHsFZwtiFROB"

    # 如果确定开发就创建应用, 根据不同的状态执行不同的任务
    if action_data["status"] == "optQgHmA8z":
        process_new_robot.delay()
    else:
        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