feishu_api_utils.py 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143
  1. import httpx
  2. from extensions.ext_redis import redis_client
  3. class FeishuRequest:
  4. def __init__(self, app_id: str, app_secret: str):
  5. self.app_id = app_id
  6. self.app_secret = app_secret
  7. @property
  8. def tenant_access_token(self):
  9. feishu_tenant_access_token = f"tools:{self.app_id}:feishu_tenant_access_token"
  10. if redis_client.exists(feishu_tenant_access_token):
  11. return redis_client.get(feishu_tenant_access_token).decode()
  12. res = self.get_tenant_access_token(self.app_id, self.app_secret)
  13. redis_client.setex(feishu_tenant_access_token, res.get("expire"), res.get("tenant_access_token"))
  14. return res.get("tenant_access_token")
  15. def _send_request(self, url: str, method: str = "post", require_token: bool = True, payload: dict = None,
  16. params: dict = None):
  17. headers = {
  18. "Content-Type": "application/json",
  19. "user-agent": "Dify",
  20. }
  21. if require_token:
  22. headers["tenant-access-token"] = f"{self.tenant_access_token}"
  23. res = httpx.request(method=method, url=url, headers=headers, json=payload, params=params, timeout=30).json()
  24. if res.get("code") != 0:
  25. raise Exception(res)
  26. return res
  27. def get_tenant_access_token(self, app_id: str, app_secret: str) -> dict:
  28. """
  29. API url: https://open.feishu.cn/document/server-docs/authentication-management/access-token/tenant_access_token_internal
  30. Example Response:
  31. {
  32. "code": 0,
  33. "msg": "ok",
  34. "tenant_access_token": "t-caecc734c2e3328a62489fe0648c4b98779515d3",
  35. "expire": 7200
  36. }
  37. """
  38. url = "https://lark-plugin-api.solutionsuite.cn/lark-plugin/access_token/get_tenant_access_token"
  39. payload = {
  40. "app_id": app_id,
  41. "app_secret": app_secret
  42. }
  43. res = self._send_request(url, require_token=False, payload=payload)
  44. return res
  45. def create_document(self, title: str, content: str, folder_token: str) -> dict:
  46. """
  47. API url: https://open.larkoffice.com/document/server-docs/docs/docs/docx-v1/document/create
  48. Example Response:
  49. {
  50. "data": {
  51. "title": "title",
  52. "url": "https://svi136aogf123.feishu.cn/docx/VWbvd4fEdoW0WSxaY1McQTz8n7d",
  53. "type": "docx",
  54. "token": "VWbvd4fEdoW0WSxaY1McQTz8n7d"
  55. },
  56. "log_id": "021721281231575fdbddc0200ff00060a9258ec0000103df61b5d",
  57. "code": 0,
  58. "msg": "创建飞书文档成功,请查看"
  59. }
  60. """
  61. url = "https://lark-plugin-api.solutionsuite.cn/lark-plugin/document/create_document"
  62. payload = {
  63. "title": title,
  64. "content": content,
  65. "folder_token": folder_token,
  66. }
  67. res = self._send_request(url, payload=payload)
  68. return res.get("data")
  69. def write_document(self, document_id: str, content: str, position: str = "start") -> dict:
  70. url = "https://lark-plugin-api.solutionsuite.cn/lark-plugin/document/write_document"
  71. payload = {
  72. "document_id": document_id,
  73. "content": content,
  74. "position": position
  75. }
  76. res = self._send_request(url, payload=payload)
  77. return res.get("data")
  78. def get_document_raw_content(self, document_id: str) -> dict:
  79. """
  80. API url: https://open.larkoffice.com/document/server-docs/docs/docs/docx-v1/document/raw_content
  81. Example Response:
  82. {
  83. "code": 0,
  84. "msg": "success",
  85. "data": {
  86. "content": "云文档\n多人实时协同,插入一切元素。不仅是在线文档,更是强大的创作和互动工具\n云文档:专为协作而生\n"
  87. }
  88. }
  89. """
  90. params = {
  91. "document_id": document_id,
  92. }
  93. url = "https://lark-plugin-api.solutionsuite.cn/lark-plugin/document/get_document_raw_content"
  94. res = self._send_request(url, method="get", params=params)
  95. return res.get("data").get("content")
  96. def list_document_block(self, document_id: str, page_token: str, page_size: int = 500) -> dict:
  97. """
  98. API url: https://open.larkoffice.com/document/server-docs/docs/docs/docx-v1/document/list
  99. """
  100. url = "https://lark-plugin-api.solutionsuite.cn/lark-plugin/document/list_document_block"
  101. params = {
  102. "document_id": document_id,
  103. "page_size": page_size,
  104. "page_token": page_token,
  105. }
  106. res = self._send_request(url, method="get", params=params)
  107. return res.get("data")
  108. def send_bot_message(self, receive_id_type: str, receive_id: str, msg_type: str, content: str) -> dict:
  109. """
  110. API url: https://open.larkoffice.com/document/server-docs/im-v1/message/create
  111. """
  112. url = "https://lark-plugin-api.solutionsuite.cn/lark-plugin/message/send_bot_message"
  113. params = {
  114. "receive_id_type": receive_id_type,
  115. }
  116. payload = {
  117. "receive_id": receive_id,
  118. "msg_type": msg_type,
  119. "content": content,
  120. }
  121. res = self._send_request(url, params=params, payload=payload)
  122. return res.get("data")
  123. def send_webhook_message(self, webhook: str, msg_type: str, content: str) -> dict:
  124. url = "https://lark-plugin-api.solutionsuite.cn/lark-plugin/message/send_webhook_message"
  125. payload = {
  126. "webhook": webhook,
  127. "msg_type": msg_type,
  128. "content": content,
  129. }
  130. res = self._send_request(url, require_token=False, payload=payload)
  131. return res