import json import os import http import requests import lark_oapi as lark from dotenv import load_dotenv import lark_oapi as lark from lark_oapi.api.docx.v1 import * import lark_oapi as lark from lark_oapi.api.contact.v3 import * from lark_oapi.api.wiki.v2 import * from lark_oapi.api.im.v1 import * def search_department(client): request: SearchDepartmentRequest = ( SearchDepartmentRequest.builder() .page_size(20) .request_body(SearchDepartmentRequestBody.builder().query("内容").build()) .build() ) option = ( lark.RequestOption.builder() .user_access_token("u-ddet2pz9R4ApKdl0ta.vAj1l6H_5ghX3h200ghK82KaD") .build() ) response: SearchDepartmentResponse = client.contact.v3.department.search( request, option ) if not response.success(): lark.logger.error( f"client.contact.v3.department.search failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}" ) return lark.logger.info(lark.JSON.marshal(response.data, indent=4)) def get_access_token(app_id, app_secret): url = "open.feishu.cn" api = "/open-apis/auth/v3/app_access_token/internal" payload = json.dumps({"app_id": app_id, "app_secret": app_secret}) headers = {"content-type": "application/json; charset=utf-8"} conn = http.client.HTTPSConnection(url) conn.request("POST", api, payload, headers) token = json.loads(conn.getresponse().read()).get("app_access_token") assert token != None, "请检查 app_id app_secret 是否正确" return token def batch_send_message(token): host = "open.feishu.cn" url = "/open-apis/message/v4/batch_send/" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", } # open_ids payload = { "msg_type": {"card": {...}}, "content": "", "user_ids": [""], } conn = http.client.HTTPSConnection(host) conn.request("POST", url, json.dumps(payload), headers) res = conn.getresponse().read() res = json.loads(res) print(res) def create_doc(client, space_id, parent_node_token, title): # client = lark.Client.builder() \ # .app_id("YOUR_APP_ID") \ # .app_secret("YOUR_APP_SECRET") \ # .log_level(lark.LogLevel.DEBUG) \ # .build() # 构造请求对象 request: CreateSpaceNodeRequest = ( CreateSpaceNodeRequest.builder() .space_id(space_id) .request_body( Node.builder() .obj_type("docx") .parent_node_token(parent_node_token) .node_type("origin") .title(title) .build() ) .build() ) # 发起请求 option = ( lark.RequestOption.builder() .user_access_token("u-drqxH7LHd2AECUzpR7iaIyh42SzMh0Dxgww0lgA024O4") .build() ) response: CreateSpaceNodeResponse = client.wiki.v2.space_node.create( request, option ) # 处理失败返回 if not response.success(): lark.logger.error( f"client.wiki.v2.space_node.create failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}" ) return print(lark.JSON.marshal(response.data)) return response.data.node.node_token, response.data.node.obj_token def get_doc_block(client, document_id): # 创建client # 使用 user_access_token 需开启 token 配置, 并在 request_option 中配置 token # 构造请求对象 request: ListDocumentBlockRequest = ( ListDocumentBlockRequest.builder() .document_id(document_id) .page_size(500) .document_revision_id(-1) .build() ) # 发起请求 option = ( lark.RequestOption.builder() .user_access_token("u-drqxH7LHd2AECUzpR7iaIyh42SzMh0Dxgww0lgA024O4") .build() ) response: ListDocumentBlockResponse = client.docx.v1.document_block.list( request, option ) # 处理失败返回 if not response.success(): lark.logger.error( f"client.docx.v1.document_block.list failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}" ) return # 处理业务结果 return response.data.items[0].block_id def create_doc_block( client, document_id, block_id, background="", features="", ): # 构造请求对象 request: CreateDocumentBlockChildrenRequest = ( CreateDocumentBlockChildrenRequest.builder() .document_id(document_id) .block_id(block_id) .document_revision_id(-1) .request_body( CreateDocumentBlockChildrenRequestBody.builder() .children( [ Block.builder() .block_type(4) .heading2( Text.builder() .elements( [ TextElement.builder() .text_run(TextRun.builder().content("需求背景").build()) .build(), ] ) .build() ) .build(), Block.builder() .block_type(2) .text( Text.builder() .style(TextStyle.builder().build()) .elements( [ TextElement.builder() .text_run(TextRun.builder().content(background).build()) .build(), ] ) .build() ) .build(), Block.builder() .block_type(4) .heading2( Text.builder() .elements( [ TextElement.builder() .text_run(TextRun.builder().content("需求功能").build()) .build(), ] ) .build() ) .build(), Block.builder() .block_type(2) .text( Text.builder() .style(TextStyle.builder().build()) .elements( [ TextElement.builder() .text_run(TextRun.builder().content(features).build()) .build(), ] ) .build() ) .build(), Block.builder() .block_type(4) .heading2( Text.builder() .elements( [ TextElement.builder() .text_run(TextRun.builder().content("需求实现").build()) .build(), ] ) .build() ) .build(), Block.builder() .block_type(4) .heading2( Text.builder() .elements( [ TextElement.builder() .text_run(TextRun.builder().content("常见问题").build()) .build(), ] ) .build() ) .build(), ] ) .index(0) .build() ) .build() ) # 发起请求 option = ( lark.RequestOption.builder() .user_access_token("u-drqxH7LHd2AECUzpR7iaIyh42SzMh0Dxgww0lgA024O4") .build() ) response: CreateDocumentBlockChildrenResponse = ( client.docx.v1.document_block_children.create(request, option) ) # 处理失败返回 if not response.success(): lark.logger.error( f"client.docx.v1.document_block_children.create failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}" ) return # 处理业务结果 lark.logger.info(lark.JSON.marshal(response.data, indent=4)) def get_messages(client, container_id_type="", container_id="", page_size=50): # 构造请求对象 request: ListMessageRequest = ( ListMessageRequest.builder() .container_id_type(container_id_type) .container_id(container_id) .sort_type("ByCreateTimeAsc") .page_size(50) .build() ) # 发起请求 response: ListMessageResponse = client.im.v1.message.list(request) # 处理失败返回 if not response.success(): lark.logger.error( f"client.im.v1.message.list failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}" ) return # 处理业务结果 lark.logger.info(lark.JSON.marshal(response.data, indent=4)) return json.loads(lark.JSON.marshal(response.data)) def send_groud_message(client, receive_id, content): # 创建client client = ( lark.Client.builder() .app_id("YOUR_APP_ID") .app_secret("YOUR_APP_SECRET") .log_level(lark.LogLevel.DEBUG) .build() ) # 构造请求对象 request: CreateMessageRequest = ( CreateMessageRequest.builder() .receive_id_type("chat_id") .request_body( CreateMessageRequestBody.builder() .receive_id(receive_id) .msg_type("text") .content(content) .build() ) .build() ) # 发起请求 response: CreateMessageResponse = client.im.v1.message.create(request) # 处理失败返回 if not response.success(): lark.logger.error( f"client.im.v1.message.create failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}, resp: \n{json.dumps(json.loads(response.raw.content), indent=4, ensure_ascii=False)}" ) return # 处理业务结果 lark.logger.info(lark.JSON.marshal(response.data, indent=4)) if __name__ == "__main__": load_dotenv() app_id = os.environ["APP_ID"] app_secret = os.environ["APP_SECRET"] data = { "title": "测试", "features": "测试", "background": "测试", "category": "opt5xydmUE", } title = data.get("title") background = data.get("background") features = data.get("features") category = data.get("category") client = ( lark.Client.builder() .app_id(app_id) .app_secret(app_secret) .log_level(lark.LogLevel.DEBUG) .build() ) space_id = "7268552249415319554" # 官方应用 if category == "opt5xydmUE": parent_node_token = "SvWCwlWJ8iuFKpkXp1rcd9u2n3c" if category == "optRTwYcra": parent_node_token = "F4ivwKrgyiQ4zOkuLVFcTq9Bnee" create_doc(client, space_id, parent_node_token, title) pass # client = ( # lark.Client.builder() # .app_id(app_id) # .app_secret(app_secret) # .log_level(lark.LogLevel.DEBUG) # .build() # ) # detail_url = ( # "https://ying-dao.feishu.cn/wiki/EdHQwOhV9iLuh7kZTkwcX6OfnIe?from=from_copylink" # ) # create_demand_url = ( # "https://ying-dao.feishu.cn/share/base/form/shrcnEI7Hhnhfdh0OXTyQcwGzEb" # ) # app_list = [ # "精选联盟批量邀约: 该在精选联盟设置筛选条件后并对达人批量邀约的功能", # "快手达人批量邀约: 在快手设置筛选条件后并对达人批量邀约的功能", # "视频号达人批量邀约: 在视频后台设置筛选条件并对达人批量邀约的功能", # ] # activity_list = [ # "批量数据抓取(Java): Java单次表格数据抓取,Java多页连续表格数据抓取", # "网页扩展操作: 智能日期选择器,滚动元素至可视区域,隐藏元素,显示元素...", # "PostgreSQL 扩展: 执行sql语句,建立连接,添加数据记录,批量添加记录...", # "天猫字体解密扩展: 获取天猫价格解密字典, 解密天猫价格内容", # ] # # app_list = [ # # "企微群发消息: 在文件传输助手中根据群名单逐条转发信息", # # "TEMU商品复制上架: TEMU商品多店铺一键铺货", # # "Instagram关键词抓取红人信息: 根据条件抓取红人邮箱以及其他基本信息", # # "TikTok关键词抓取红人信息: 根据条件筛选符合条件的红人，获取红人信息", # # "查集装箱实际到港时间-6大航司: 根据船公司和提单号在对应的官网查询到港时间", # # ] # # activity_list = [ # # "云码验证码扩展: 通用轨迹验证, 通用单图滑块验证, 京东-旋转验证码", # # "Excel 扩展操作: 设置筛选器", # # ] # app_text = "\n".join(f"🔹{item}" for item in app_list).strip() # activity_text = "\n".join(f"🔸{item}" for item in activity_list).strip() # # 解决方案实施中心 4gge82cgb18f52af # # 电商事业部 8a612b6c9184b118 # # 技术支持部 d7g1374b1238d267 # # 杭州内容组 9fg37g65b1b5b24e # # Uerid 7g3756f4 # # "user_ids": [ # # "7g3756f4", # # ], # pyload = { # "department_ids": ["4gge82cgb18f52af", "8a612b6c9184b118", "d7g1374b1238d267"], # "user_ids": [], # "msg_type": "interactive", # "card": { # "type": "template", # "data": { # "template_id": "ctp_AAikuuqPwN0o", # "template_variable": { # "app_text": app_text, # "activity_text": activity_text, # "detail_url": { # "pc_url": "", # "android_url": "", # "ios_url": "", # "url": detail_url, # }, # "create_demand_url": { # "pc_url": "", # "android_url": "", # "ios_url": "", # "url": create_demand_url, # }, # }, # }, # }, # } # # 构造请求对象 # request: lark.BaseRequest = ( # lark.BaseRequest.builder() # .http_method(lark.HttpMethod.POST) # .uri("/open-apis/message/v4/batch_send/") # .token_types({lark.AccessTokenType.TENANT}) # .body(pyload) # .build() # ) # # 发起请求 # response: lark.BaseResponse = client.request(request) # # 处理失败返回 # if not response.success(): # lark.logger.error( # f"client.request failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}" # ) # # 处理业务结果 # lark.logger.info(str(response.raw.content, lark.UTF_8))