import json

import time

from typing import Any, Optional, cast, Union, Dict

import httpx

from requests_toolbelt import MultipartEncoder

# from core.tools.errors import ToolProviderCredentialValidationError

# from extensions.ext_redis import redis_client

# 新增导入 tenacity 与 random

import tenacity # 用于重试机制

import random # 用于生成随机等待时间

import logging

from .message import MessageAPI

from .contact import ContactAPI

from .chat import ChatAPI # 添加导入

from .retry import custom_retry, custom_wait

from .bitable import BitableAPI # 添加导入

from .wiki import WikiAPI # 添加导入

from .drive import DriveAPI # 添加导入

class ToolProviderCredentialValidationError(ValueError):

pass

def auth(credentials):

app_id = credentials.get("app_id")

app_secret = credentials.get("app_secret")

if not app_id or not app_secret:

raise ToolProviderCredentialValidationError("app_id and app_secret is required")

try:

assert FeishuRequest(app_id, app_secret).tenant_access_token is not None

except Exception as e:

raise ToolProviderCredentialValidationError(str(e))

def convert_add_records(json_str):

try:

data = json.loads(json_str)

if not isinstance(data, list):

raise ValueError("Parsed data must be a list")

converted_data = [{"fields": json.dumps(item, ensure_ascii=False)} for item in data]

return converted_data

except json.JSONDecodeError:

raise ValueError("The input string is not valid JSON")

except Exception as e:

raise ValueError(f"An error occurred while processing the data: {e}")

def convert_update_records(json_str):

try:

data = json.loads(json_str)

if not isinstance(data, list):

raise ValueError("Parsed data must be a list")

converted_data = [

{"fields": json.dumps(record["fields"], ensure_ascii=False), "record_id": record["record_id"]}

for record in data

if "fields" in record and "record_id" in record

]

if len(converted_data) != len(data):

raise ValueError("Each record must contain 'fields' and 'record_id'")

return converted_data

except json.JSONDecodeError:

raise ValueError("The input string is not valid JSON")

except Exception as e:

raise ValueError(f"An error occurred while processing the data: {e}")

class FeishuRequest:

# API_BASE_URL = "https://lark-plugin-api.solutionsuite.cn/lark-plugin"

API_BASE_URL = "https://open.work.sany.com.cn/open-apis"

def __init__(self, app_id: str = None, app_secret: str = None, tenant_access_token: str = None, base_url: str = API_BASE_URL):

self.logger = logging.getLogger(__name__)

self.logger.debug(

f"Creating FeishuRequest app_id: {app_id}, app_secret: {app_secret}, tenant_access_token: {tenant_access_token}, base_url: {base_url}"

)

self.app_id = app_id

self.app_secret = app_secret

self.API_BASE_URL = base_url

self._tenant_access_token = tenant_access_token

self._token_expire_time = 0 if not tenant_access_token else time.time() + 3

self.message = MessageAPI(self)

self.contact = ContactAPI(self)

self.chat = ChatAPI(self)

self.bitable = BitableAPI(self) # 添加初始化

self.wiki = WikiAPI(self)

self.drive = DriveAPI(self)

@property

def tenant_access_token(self):

# # dify 中的原始代码如下

# feishu_tenant_access_token = f"tools:{self.app_id}:feishu_tenant_access_token"

# if redis_client.exists(feishu_tenant_access_token):

# return redis_client.get(feishu_tenant_access_token).decode()

# res = self.get_tenant_access_token(self.app_id, self.app_secret)

# redis_client.setex(feishu_tenant_access_token, res.get("expire"), res.get("tenant_access_token"))

# return res.get("tenant_access_token")

"""获取租户访问令牌

如果令牌不存在或已过期，则重新获取

Returns:

tenant_access_token: 租户访问令牌

"""

should_refresh = not self._tenant_access_token or time.time() >= self._token_expire_time

if should_refresh:

print(f"刷新租户访问令牌: {'令牌不存在' if not self._tenant_access_token else '令牌过期'}")

self._refresh_tenant_access_token()

return self._tenant_access_token

def _refresh_tenant_access_token(self) -> None:

"""刷新租户访问令牌"""

url = f"{self.API_BASE_URL}/auth/v3/tenant_access_token/internal"

payload = {

"app_id": self.app_id,

"app_secret": self.app_secret,

}

res = self.get_tenant_access_token(self.app_id, self.app_secret)

self._tenant_access_token = res.get("tenant_access_token")

self._token_expire_time = time.time() + res.get("expire") - 60 # 提前60秒刷新

@tenacity.retry(

retry=custom_retry,

wait=custom_wait,

stop=tenacity.stop_after_attempt(3),

reraise=True,

)

def _send_request(

self,

url: str,

method: str = "post",

require_token: bool = True,

payload: Optional[dict] = None,

params: Optional[dict] = None,

response_stream: bool = False,

multi_form: Optional[MultipartEncoder] = None,

) -> Union[dict, bytes]:

"""发送请求到飞书API

Args:

url: 请求URL

method: 请求方法，默认为post

require_token: 是否需要认证token

payload: 请求体数据

params: URL参数

response_stream: 是否返回二进制流数据

multi_form: 多部分表单数据,如果不为空，说明使用multipart/form-data方式上传文件

Returns:

Union[dict, bytes]: 如果response_stream=True返回二进制数据，否则返回JSON响应

"""

if multi_form:

headers = {

"Content-Type": multi_form.content_type,

"user-agent": "Dify",

}

else:

headers = {

"Content-Type": "application/json",

"user-agent": "Dify",

}

if require_token:

headers["tenant-access-token"] = f"{self.tenant_access_token}"

headers["Authorization"] = f"Bearer {self.tenant_access_token}"

# Generate curl command for debugging

curl_command = f"curl -X {method.upper()} '{url}'"

for header, value in headers.items():

curl_command += f" -H '{header}: {value}'"

if payload:

curl_command += f" -d '{json.dumps(payload)}'"

if params:

params_str = "&".join([f"{k}={v}" for k, v in params.items()])

curl_command += f" -G --data-urlencode '{params_str}'"

print(f"\nDebug curl command:\n{curl_command}\n")

if multi_form:

response = httpx.request(method=method, url=url, headers=headers, data=multi_form.read(), timeout=30)

else:

response = httpx.request(method=method, url=url, headers=headers, json=payload, params=params, timeout=30)

# 如果返回content-type包含application/json，说明接口未能正常返回二进制流

if response_stream and "application/json" not in response.headers.get("Content-Type"):

return response.content

# 否则按原来的方式处理JSON响应

feishu_response = response.json()

ratelimit_reset = response.headers.get("x-ogw-ratelimit-reset")

ratelimit_limit = response.headers.get("x-ogw-ratelimit-limit")

if feishu_response.get("code") != 0:

if ratelimit_reset:

feishu_response.update({"ratelimit_reset": ratelimit_reset})

if ratelimit_limit:

feishu_response.update({"ratelimit_limit": ratelimit_limit})

feishu_response.update({"http_status": response.status_code})

raise Exception(feishu_response)

return feishu_response

def get_tenant_access_token(self, app_id: str, app_secret: str) -> dict:

"""

API url: https://open.feishu.cn/document/server-docs/authentication-management/access-token/tenant_access_token_internal

Example Response:

{

"code": 0,

"msg": "ok",

"tenant_access_token": "t-caecc734c2e3328a62489fe0648c4b98779515d3",

"expire": 7200

}

"""

# url = f"{self.API_BASE_URL}/access_token/get_tenant_access_token"

url = f"{self.API_BASE_URL}/auth/v3/tenant_access_token/internal"

payload = {"app_id": app_id, "app_secret": app_secret}

res: dict = self._send_request(url, require_token=False, payload=payload)

return res

def create_document(self, title: str, content: str, folder_token: str) -> dict:

"""

API url: https://open.larkoffice.com/document/server-docs/docs/docs/docx-v1/document/create

Example Response:

{

"data": {

"title": "title",

"url": "https://svi136aogf123.feishu.cn/docx/VWbvd4fEdoW0WSxaY1McQTz8n7d",

"type": "docx",

"token": "VWbvd4fEdoW0WSxaY1McQTz8n7d"

},

"log_id": "021721281231575fdbddc0200ff00060a9258ec0000103df61b5d",

"code": 0,

"msg": "创建飞书文档成功，请查看"

}

"""

url = f"{self.API_BASE_URL}/document/create_document"

payload = {

"title": title,

"content": content,

"folder_token": folder_token,

}

res: dict = self._send_request(url, payload=payload)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def write_document(self, document_id: str, content: str, position: str = "end") -> dict:

url = f"{self.API_BASE_URL}/document/write_document"

payload = {"document_id": document_id, "content": content, "position": position}

res: dict = self._send_request(url, payload=payload)

return res

def get_document_content(self, document_id: str, mode: str = "markdown", lang: str = "0") -> str:

"""

API url: https://open.larkoffice.com/document/server-docs/docs/docs/docx-v1/document/raw_content

Example Response:

{

"code": 0,

"msg": "success",

"data": {

"content": "云文档\n多人实时协同，插入一切元素。不仅是在线文档，更是强大的创作和互动工具\n云文档：专为协作而生\n"

}

}

""" # noqa: E501

params = {

"document_id": document_id,

"mode": mode,

"lang": lang,

}

url = f"{self.API_BASE_URL}/document/get_document_content"

res: dict = self._send_request(url, method="GET", params=params)

if "data" in res:

return cast(str, res.get("data", {}).get("content"))

return ""

def list_document_blocks(self, document_id: str, page_token: str, user_id_type: str = "open_id", page_size: int = 500) -> dict:

"""

API url: https://open.larkoffice.com/document/server-docs/docs/docs/docx-v1/document/list

"""

params = {

"user_id_type": user_id_type,

"document_id": document_id,

"page_size": page_size,

"page_token": page_token,

}

url = f"{self.API_BASE_URL}/document/list_document_blocks"

res: dict = self._send_request(url, method="GET", params=params)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def get_chat_messages(

self,

container_id: str,

start_time: str = None,

end_time: str = None,

page_token: str = None,

sort_type: str = "ByCreateTimeAsc",

page_size: int = 20,

) -> dict:

# # dify 中的原始代码如下

# """

# API url: https://open.larkoffice.com/document/server-docs/im-v1/message/list

# """

# url = f"{self.API_BASE_URL}/message/get_chat_messages"

# params = {

# "container_id": container_id,

# "start_time": start_time,

# "end_time": end_time,

# "sort_type": sort_type,

# "page_token": page_token,

# "page_size": page_size,

# }

# res: dict = self._send_request(url, method="GET", params=params)

# if "data" in res:

# data: dict = res.get("data", {})

# return data

# return res

return self.get_history_messages(

container_id_type="chat",

container_id=container_id,

page_size=page_size,

page_token=page_token,

sort_type=sort_type,

)

def get_thread_messages(self, container_id: str, page_token: str = None, sort_type: str = "ByCreateTimeAsc", page_size: int = 20) -> dict:

# # dify 中的原始代码如下

# """

# API url: https://open.larkoffice.com/document/server-docs/im-v1/message/list

# """

# url = f"{self.API_BASE_URL}/message/get_thread_messages"

# params = {

# "container_id": container_id,

# "sort_type": sort_type,

# "page_token": page_token,

# "page_size": page_size,

# }

# res: dict = self._send_request(url, method="GET", params=params)

# if "data" in res:

# data: dict = res.get("data", {})

# return data

# return res

return self.get_history_messages(

container_id_type="thread",

container_id=container_id,

page_size=page_size,

page_token=page_token,

sort_type=sort_type,

)

def get_history_messages(

self,

container_id_type: str = "chat",

container_id: str = None,

page_size: int = 20,

page_token: str = None,

sort_type: str = None,

start_time: int = None,

end_time: int = None,

) -> dict:

"""获取会话历史消息

Args:

container_type: 会话类型，可选值有：chat(包含单聊p2p和群聊group)，thread（话题）

container_id: 会话ID，如果是群聊，则为群聊的chat_id

page_size: 分页大小，默认20，最大100

page_token: 分页标记，第一次请求不填，表示从最新的消息开始翻页

sort_type: 排序类型，可选值有：ByCreateTimeAsc（按创建时间升序）、ByCreateTimeDesc（按创建时间降序）

start_time: 开始时间，毫秒级时间戳

end_time: 结束时间，毫秒级时间戳

Returns:

Dict: 包含消息列表的响应数据

API文档: https://open.feishu.cn/document/uAjLw4CM/ukTMukTMukTM/reference/im-v1/message/list

"""

url = f"{self.API_BASE_URL}/im/v1/messages"

params = {

"container_id_type": container_id_type,

"container_id": container_id,

"page_size": page_size,

}

if page_token:

params["page_token"] = page_token

if sort_type:

params["sort_type"] = sort_type

if start_time:

params["start_time"] = start_time

if end_time:

params["end_time"] = end_time

res: dict = self._send_request(url, method="GET", params=params)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def create_task(self, summary: str, start_time: str, end_time: str, completed_time: str, description: str) -> dict:

# 创建任务

url = f"{self.API_BASE_URL}/task/create_task"

payload = {

"summary": summary,

"start_time": start_time,

"end_time": end_time,

"completed_at": completed_time,

"description": description,

}

res: dict = self._send_request(url, payload=payload)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def update_task(self, task_guid: str, summary: str, start_time: str, end_time: str, completed_time: str, description: str) -> dict:

# 更新任务

url = f"{self.API_BASE_URL}/task/update_task"

payload = {

"task_guid": task_guid,

"summary": summary,

"start_time": start_time,

"end_time": end_time,

"completed_time": completed_time,

"description": description,

}

res: dict = self._send_request(url, method="PATCH", payload=payload)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def delete_task(self, task_guid: str) -> dict:

# 删除任务

url = f"{self.API_BASE_URL}/task/delete_task"

payload = {

"task_guid": task_guid,

}

res: dict = self._send_request(url, method="DELETE", payload=payload)

return res

def add_members(self, task_guid: str, member_phone_or_email: str, member_role: str) -> dict:

# 删除任务

url = f"{self.API_BASE_URL}/task/add_members"

payload = {

"task_guid": task_guid,

"member_phone_or_email": member_phone_or_email,

"member_role": member_role,

}

res: dict = self._send_request(url, payload=payload)

return res

def get_wiki_nodes(self, space_id: str, parent_node_token: str, page_token: str, page_size: int = 20) -> dict:

# 获取知识库全部子节点列表

url = f"{self.API_BASE_URL}/wiki/get_wiki_nodes"

payload = {

"space_id": space_id,

"parent_node_token": parent_node_token,

"page_token": page_token,

"page_size": page_size,

}

res: dict = self._send_request(url, payload=payload)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def get_primary_calendar(self, user_id_type: str = "open_id") -> dict:

url = f"{self.API_BASE_URL}/calendar/get_primary_calendar"

params = {

"user_id_type": user_id_type,

}

res: dict = self._send_request(url, method="GET", params=params)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def create_event(

self,

summary: str,

description: str,

start_time: str,

end_time: str,

attendee_ability: str,

need_notification: bool = True,

auto_record: bool = False,

) -> dict:

url = f"{self.API_BASE_URL}/calendar/create_event"

payload = {

"summary": summary,

"description": description,

"need_notification": need_notification,

"start_time": start_time,

"end_time": end_time,

"auto_record": auto_record,

"attendee_ability": attendee_ability,

}

res: dict = self._send_request(url, payload=payload)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def update_event(

self,

event_id: str,

summary: str,

description: str,

need_notification: bool,

start_time: str,

end_time: str,

auto_record: bool,

) -> dict:

url = f"{self.API_BASE_URL}/calendar/update_event/{event_id}"

payload: dict[str, Any] = {}

if summary:

payload["summary"] = summary

if description:

payload["description"] = description

if start_time:

payload["start_time"] = start_time

if end_time:

payload["end_time"] = end_time

if need_notification:

payload["need_notification"] = need_notification

if auto_record:

payload["auto_record"] = auto_record

res: dict = self._send_request(url, method="PATCH", payload=payload)

return res

def delete_event(self, event_id: str, need_notification: bool = True) -> dict:

url = f"{self.API_BASE_URL}/calendar/delete_event/{event_id}"

params = {

"need_notification": need_notification,

}

res: dict = self._send_request(url, method="DELETE", params=params)

return res

def list_events(self, start_time: str, end_time: str, page_token: str, page_size: int = 50) -> dict:

url = f"{self.API_BASE_URL}/calendar/list_events"

params = {

"start_time": start_time,

"end_time": end_time,

"page_token": page_token,

"page_size": page_size,

}

res: dict = self._send_request(url, method="GET", params=params)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def search_events(

self,

query: str,

start_time: str,

end_time: str,

page_token: str,

user_id_type: str = "open_id",

page_size: int = 20,

) -> dict:

url = f"{self.API_BASE_URL}/calendar/search_events"

payload = {

"query": query,

"start_time": start_time,

"end_time": end_time,

"page_token": page_token,

"user_id_type": user_id_type,

"page_size": page_size,

}

res: dict = self._send_request(url, payload=payload)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def add_event_attendees(self, event_id: str, attendee_phone_or_email: str, need_notification: bool = True) -> dict:

# 参加日程参会人

url = f"{self.API_BASE_URL}/calendar/add_event_attendees"

payload = {

"event_id": event_id,

"attendee_phone_or_email": attendee_phone_or_email,

"need_notification": need_notification,

}

res: dict = self._send_request(url, payload=payload)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def create_spreadsheet(

self,

title: str,

folder_token: str,

) -> dict:

# 创建电子表格

url = f"{self.API_BASE_URL}/spreadsheet/create_spreadsheet"

payload = {

"title": title,

"folder_token": folder_token,

}

res: dict = self._send_request(url, payload=payload)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def get_spreadsheet(

self,

spreadsheet_token: str,

user_id_type: str = "open_id",

) -> dict:

# 获取电子表格信息

url = f"{self.API_BASE_URL}/spreadsheet/get_spreadsheet"

params = {

"spreadsheet_token": spreadsheet_token,

"user_id_type": user_id_type,

}

res: dict = self._send_request(url, method="GET", params=params)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def list_spreadsheet_sheets(

self,

spreadsheet_token: str,

) -> dict:

# 列出电子表格的所有工作表

url = f"{self.API_BASE_URL}/spreadsheet/list_spreadsheet_sheets"

params = {

"spreadsheet_token": spreadsheet_token,

}

res: dict = self._send_request(url, method="GET", params=params)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def add_rows(

self,

spreadsheet_token: str,

sheet_id: str,

sheet_name: str,

length: int,

values: str,

) -> dict:

# 增加行,在工作表最后添加

url = f"{self.API_BASE_URL}/spreadsheet/add_rows"

payload = {

"spreadsheet_token": spreadsheet_token,

"sheet_id": sheet_id,

"sheet_name": sheet_name,

"length": length,

"values": values,

}

res: dict = self._send_request(url, payload=payload)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def add_cols(

self,

spreadsheet_token: str,

sheet_id: str,

sheet_name: str,

length: int,

values: str,

) -> dict:

# 增加列,在工作表最后添加

url = f"{self.API_BASE_URL}/spreadsheet/add_cols"

payload = {

"spreadsheet_token": spreadsheet_token,

"sheet_id": sheet_id,

"sheet_name": sheet_name,

"length": length,

"values": values,

}

res: dict = self._send_request(url, payload=payload)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def read_rows(

self,

spreadsheet_token: str,

sheet_id: str,

sheet_name: str,

start_row: int,

num_rows: int,

user_id_type: str = "open_id",

) -> dict:

# 读取工作表行数据

url = f"{self.API_BASE_URL}/spreadsheet/read_rows"

params = {

"spreadsheet_token": spreadsheet_token,

"sheet_id": sheet_id,

"sheet_name": sheet_name,

"start_row": start_row,

"num_rows": num_rows,

"user_id_type": user_id_type,

}

res: dict = self._send_request(url, method="GET", params=params)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def read_cols(

self,

spreadsheet_token: str,

sheet_id: str,

sheet_name: str,

start_col: int,

num_cols: int,

user_id_type: str = "open_id",

) -> dict:

# 读取工作表列数据

url = f"{self.API_BASE_URL}/spreadsheet/read_cols"

params = {

"spreadsheet_token": spreadsheet_token,

"sheet_id": sheet_id,

"sheet_name": sheet_name,

"start_col": start_col,

"num_cols": num_cols,

"user_id_type": user_id_type,

}

res: dict = self._send_request(url, method="GET", params=params)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def read_table(

self,

spreadsheet_token: str,

sheet_id: str,

sheet_name: str,

num_range: str,

query: str,

user_id_type: str = "open_id",

) -> dict:

# 自定义读取行列数据

url = f"{self.API_BASE_URL}/spreadsheet/read_table"

params = {

"spreadsheet_token": spreadsheet_token,

"sheet_id": sheet_id,

"sheet_name": sheet_name,

"range": num_range,

"query": query,

"user_id_type": user_id_type,

}

res: dict = self._send_request(url, method="GET", params=params)

if "data" in res:

data: dict = res.get("data", {})

return data

return res

def __getattr__(self, name):

"""自动代理到各个API模块的方法"""

self.logger.debug(f"Attempting to access attribute '{name}'")

if hasattr(self.contact, name):

self.logger.debug(f"Attribute '{name}' found in 'contact'")

return getattr(self.contact, name)

if hasattr(self.message, name):

self.logger.debug(f"Attribute '{name}' found in 'message'")

return getattr(self.message, name)

if hasattr(self.chat, name):

self.logger.debug(f"Attribute '{name}' found in 'chat'")

return getattr(self.chat, name)

if hasattr(self.bitable, name): # 添加 bitable 的方法代理

self.logger.debug(f"Attribute '{name}' found in 'bitable'")

return getattr(self.bitable, name)

if hasattr(self.wiki, name):

self.logger.debug(f"Attribute '{name}' found in 'wiki'")

return getattr(self.wiki, name)

if hasattr(self.drive, name):

self.logger.debug(f"Attribute '{name}' found in 'drive'")

return getattr(self.drive, name)

self.logger.debug(f"Attribute '{name}' not found in any API module")

raise AttributeError(f"'FeishuRequest' object has no attribute '{name}'")

def extract_user_info(self, user_data):

"""

从 user_data 中提取用户部门信息和职位信息

Args:

user_data: 包含用户信息的字典

Returns:

dict: 包含部门和职位信息的字典

"""

# 初始化结果

result = {

"open_id": user_data.get("open_id", ""),

"name": user_data.get("name", ""),

"en_name": user_data.get("en_name", ""),

"email": user_data.get("email", ""),

"user_id": user_data.get("user_id", ""),

"department": None,

"position": None,

}

# 确保存在用户信息和自定义属性

if "custom_attrs" not in user_data:

return result

# 遍历自定义属性

for attr in user_data["custom_attrs"]:

# 提取部门信息 (ID为C-6875502414099644537)

if attr["id"] == "C-6875502414099644537" and attr["type"] == "TEXT":

result["department"] = attr["value"]["text"]

# 提取职位信息 (ID为C-6857485831817396345)

elif attr["id"] == "C-6857485831817396345" and attr["type"] == "TEXT":

result["position"] = attr["value"]["text"]

return result

def get_message_full_detail(self, message_id: str) -> dict:

"""获取单条消息的完整详情，包括消息内容、发送者信息和聊天信息"""

message_info = self.message.get_message_info(message_id).get("items", {})[0]

self.logger.debug(f"message_info: {message_info}")

sender_id = message_info.get("sender", {}).get("id", None)

chat_id = message_info.get("chat_id", None)

if sender_id:

user_data = self.contact.get_contact_info(user_id=sender_id, user_id_type="open_id").get("user")

sender_info = self.extract_user_info(user_data)

self.logger.debug(f"sender_info: {sender_info}")

if chat_id:

chat_data = self.chat.get_chat_info(chat_id=chat_id)

chat_info = {

"name": chat_data.get("name", ""),

"chat_mode": chat_data.get("chat_mode", ""),

}

self.logger.debug(f"chat_info: {chat_data}")

return {"message_info": message_info, "sender_info": sender_info, "chat_info": chat_info}