from __future__ import annotations import lark_oapi from lark_oapi.api.im.v1 import CreateImageRequest, CreateImageRequestBody, CreateFileRequest, CreateFileRequestBody import traceback import typing import asyncio import re import base64 import uuid import json import time import datetime import hashlib from Crypto.Cipher import AES import tempfile import os import threading from langbot.pkg.utils import httpclient import lark_oapi.ws.exception import quart from lark_oapi.api.im.v1 import * import pydantic from lark_oapi.api.cardkit.v1 import * from lark_oapi.api.auth.v3 import * from lark_oapi.core.model import * import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter import langbot_plugin.api.entities.builtin.platform.message as platform_message import langbot_plugin.api.entities.builtin.platform.events as platform_events import langbot_plugin.api.entities.builtin.platform.entities as platform_entities import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger import langbot_plugin.api.entities.builtin.provider.session as provider_session _MAX_LARK_MEDIA_BYTES = 10 * 1024 * 1024 def _decode_lark_base64_limited(value: str) -> bytes: if ',' in value: value = value.split(',', 1)[1] max_encoded_bytes = 4 * ((_MAX_LARK_MEDIA_BYTES + 2) // 3) if len(value) > max_encoded_bytes: raise ValueError('Lark media exceeds the size limit') decoded = base64.b64decode(value) if len(decoded) > _MAX_LARK_MEDIA_BYTES: raise ValueError('Lark media exceeds the size limit') return decoded def _read_lark_path_limited(path: str) -> bytes: if os.path.getsize(path) > _MAX_LARK_MEDIA_BYTES: raise ValueError('Lark media exceeds the size limit') with open(path, 'rb') as file: body = file.read(_MAX_LARK_MEDIA_BYTES + 1) if len(body) > _MAX_LARK_MEDIA_BYTES: raise ValueError('Lark media exceeds the size limit') return body def _write_lark_temp_file(data: bytes) -> str: with tempfile.NamedTemporaryFile(delete=False) as temp_file: temp_file.write(data) temp_file.flush() return temp_file.name def _read_lark_response_file_limited(response) -> bytes: content_length = response.raw.headers.get('content-length') if content_length is not None: try: if int(content_length) > _MAX_LARK_MEDIA_BYTES: raise ValueError('Lark media exceeds the size limit') except (TypeError, ValueError) as exc: if 'exceeds' in str(exc): raise body = response.file.read(_MAX_LARK_MEDIA_BYTES + 1) if len(body) > _MAX_LARK_MEDIA_BYTES: raise ValueError('Lark media exceeds the size limit') return body def _lark_form_component_name(prefix: str, field_name: str, index: int) -> str: safe_name = re.sub(r'[^A-Za-z0-9_]', '_', field_name)[:8] or 'field' digest = hashlib.sha1(field_name.encode('utf-8')).hexdigest()[:6] return f'{prefix}_{index}_{safe_name}_{digest}'[:32] def _dify_field_name(field: dict) -> str: return str(field.get('output_variable_name') or field.get('name') or field.get('id') or '').strip() def _dify_field_type(field: dict) -> str: return str(field.get('type') or 'text').strip().lower() def _dify_select_options(field: dict) -> list[str]: source = field.get('option_source') or {} value = source.get('value') if isinstance(source, dict) else None if isinstance(value, list): return [str(item) for item in value] if isinstance(value, str): return [part.strip() for part in value.splitlines() if part.strip()] options = field.get('options') if isinstance(options, list): result: list[str] = [] for item in options: if isinstance(item, dict): result.append(str(item.get('label') or item.get('value') or '')) else: result.append(str(item)) return [item for item in result if item] return [] def _dify_default_value(field: dict) -> str: default = field.get('default') if isinstance(default, dict): value = default.get('value') if default.get('type') == 'constant' or 'value' in default else '' else: value = default return '' if value is None else str(value) def _lark_clean_form_content(form_content: str, input_defs: list[dict]) -> str: field_names = {_dify_field_name(field) for field in input_defs if _dify_field_name(field)} kept_lines: list[str] = [] for line in (form_content or '').splitlines(): placeholder = re.fullmatch(r'\s*\{\{#\$output\.([^#{}]+)#\}\}\s*', line) if placeholder and placeholder.group(1) in field_names: continue kept_lines.append(line) return re.sub(r'\n{3,}', '\n\n', '\n'.join(kept_lines).strip()) def _lark_form_input_defs(form_data: dict) -> list[dict]: return list(form_data.get('all_input_defs') or form_data.get('input_defs') or []) def _lark_current_input_defs(form_data: dict) -> list[dict]: """Return only the field that belongs to the current interactive step.""" if form_data.get('_action_select_only'): return [] input_defs = list(form_data.get('input_defs') or []) current_field = str(form_data.get('_current_input_field') or '').strip() if not current_field: return input_defs return [field for field in input_defs if _dify_field_name(field) == current_field] def _lark_should_update_stream_element( *, resume_from: bool, form_data: dict | None, msg_seq: int, is_final: bool, ) -> bool: """Return whether the still-open streaming element should be updated.""" return not resume_from and not form_data and (msg_seq % 8 == 0 or is_final) def _lark_final_layout_texts( *, resume_from: bool, text_message: str, pre_pause_cached: str | None, resume_cached: str, ) -> tuple[str, str]: """Return (main_text, resume_placeholder_text) for the final card update. Non-resume round: the full reply belongs in the main streaming element only — also rendering the resume placeholder duplicates the reply, since both hold the same accumulated text. Resume round (Dify HITL): keep the pre-pause text in the main element and the resumed text in the placeholder, as they are distinct segments. """ if resume_from: # An empty pre-pause cache is valid (Dify paused before emitting any # text); only a missing entry (None) falls back to the full text. main_text = text_message if pre_pause_cached is None else pre_pause_cached return main_text, resume_cached return text_message, '' def _lark_display_input_value(field: dict, value: typing.Any) -> str: field_type = _dify_field_type(field) if field_type == 'file': if isinstance(value, dict): return value.get('url') or value.get('upload_file_id') or '1 file' return str(value) if field_type == 'file-list': if isinstance(value, list): return f'{len(value)} file(s)' return str(value) if isinstance(value, dict): if 'value' in value and value.get('value') not in (None, ''): return str(value.get('value')) text = value.get('text') if isinstance(text, dict): content = text.get('content') if content not in (None, ''): return str(content) if text not in (None, ''): return str(text) if isinstance(value, list): return ', '.join(_lark_display_input_value(field, item) for item in value if item not in (None, '')) return str(value) def _lark_visible_form_content(form_data: dict) -> str: """Return stage content with completed values interleaved for final actions.""" source_content = form_data.get('form_content') or '' if form_data.get('_action_select_only'): source_content = form_data.get('raw_form_content') or source_content fields = { _dify_field_name(field): field for field in _lark_form_input_defs(form_data) if _dify_field_name(field) } inputs = form_data.get('inputs') or {} def replace_placeholder(match: re.Match[str]) -> str: field_name = match.group(1).strip() field = fields.get(field_name) if not field or inputs.get(field_name) in (None, '', []): return '' lines = _lark_completed_input_lines( { 'input_defs': [field], 'inputs': {field_name: inputs[field_name]}, } ) return lines[0] if lines else '' source_content = re.sub( r'\{\{#\$output\.([^#{}]+)#\}\}', replace_placeholder, str(source_content), ) return _lark_clean_form_content( str(source_content), _lark_form_input_defs(form_data), ) def _lark_completed_input_lines(form_data: dict) -> list[str]: inputs = form_data.get('inputs') or {} if not isinstance(inputs, dict): return [] lines: list[str] = [] for field in _lark_form_input_defs(form_data): field_name = _dify_field_name(field) if not field_name: continue value = inputs.get(field_name) if value in (None, '', []): continue display_value = _lark_display_input_value(field, value) lines.append(f'✅ {field_name}：{display_value}') return lines def _lark_mapping_from_value(value: typing.Any) -> dict: if isinstance(value, dict): return value if isinstance(value, str): try: parsed = json.loads(value) except (json.JSONDecodeError, TypeError): return {} return parsed if isinstance(parsed, dict) else {} return {} def _lark_action_attr(action: typing.Any, name: str) -> typing.Any: if isinstance(action, dict): return action.get(name) return getattr(action, name, None) def _lark_extract_action_form_inputs(action: typing.Any, action_value_obj: dict) -> dict: input_name_map = action_value_obj.get('input_name_map', {}) if not isinstance(input_name_map, dict): input_name_map = {} form_value = _lark_mapping_from_value(_lark_action_attr(action, 'form_value')) if not form_value: for key in ('form_value', 'formValue', 'form_values', 'formValues'): form_value = _lark_mapping_from_value(action_value_obj.get(key)) if form_value: break if not form_value: action_name = _lark_action_attr(action, 'name') input_value = _lark_action_attr(action, 'input_value') option_value = _lark_action_attr(action, 'option') if action_name and input_value not in (None, ''): form_value = {action_name: input_value} elif action_name and option_value not in (None, ''): form_value = {action_name: option_value} form_inputs = {} for component_name, value in form_value.items(): field_name = input_name_map.get(component_name) if not field_name and isinstance(component_name, str) and '.' in component_name: field_name = input_name_map.get(component_name.rsplit('.', 1)[-1], component_name) if not field_name: field_name = component_name if field_name and value not in (None, '', []): form_inputs[str(field_name)] = value return form_inputs class NonBlockingLarkWSClient(lark_oapi.ws.Client): """Keep the SDK's synchronous connection lookup off LangBot's event loop. lark-oapi performs ``requests.post`` inside its async ``_connect`` method. A stalled TLS handshake therefore freezes Quart and every other adapter in the process. Pre-fetch the URL in a worker thread, then let the SDK finish the WebSocket setup on the main loop with the already-resolved URL. """ def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._langbot_connect_lock = asyncio.Lock() async def _connect(self) -> None: async with self._langbot_connect_lock: if self._conn is not None: return conn_url = await asyncio.to_thread(self._get_conn_url) original_get_conn_url = self._get_conn_url self._get_conn_url = lambda: conn_url try: await super()._connect() finally: self._get_conn_url = original_get_conn_url class AESCipher(object): def __init__(self, key): self.bs = AES.block_size self.key = hashlib.sha256(AESCipher.str_to_bytes(key)).digest() @staticmethod def str_to_bytes(data): u_type = type(b''.decode('utf8')) if isinstance(data, u_type): return data.encode('utf8') return data @staticmethod def _unpad(s): return s[: -ord(s[len(s) - 1 :])] def decrypt(self, enc): iv = enc[: AES.block_size] cipher = AES.new(self.key, AES.MODE_CBC, iv) return self._unpad(cipher.decrypt(enc[AES.block_size :])) def decrypt_string(self, enc): enc = base64.b64decode(enc) return self.decrypt(enc).decode('utf8') class LarkMessageConverter(abstract_platform_adapter.AbstractMessageConverter): @staticmethod async def upload_image_to_lark(msg: platform_message.Image, api_client: lark_oapi.Client) -> typing.Optional[str]: """Upload an image to Lark and return the image_key, or None if upload fails.""" try: image_bytes, _mime_type = await msg.get_bytes() except Exception as exc: print(f'Failed to load Lark image: {exc}') traceback.print_exc() return None if not image_bytes: print( f'No image data available for Image message (url={msg.url}, base64={bool(msg.base64)}, path={msg.path})' ) return None try: temp_file_path = await asyncio.to_thread( _write_lark_temp_file, image_bytes, ) try: with open(temp_file_path, 'rb') as upload_file: request = ( CreateImageRequest.builder() .request_body(CreateImageRequestBody.builder().image_type('message').image(upload_file).build()) .build() ) response = await api_client.im.v1.image.acreate(request) if not response.success(): print( f'client.im.v1.image.create failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}' ) return None return response.data.image_key finally: # Clean up the temporary file os.unlink(temp_file_path) except Exception as e: print(f'Failed to upload image to Lark: {e}') traceback.print_exc() return None @staticmethod async def upload_file_to_lark( file_bytes: bytes, api_client: lark_oapi.Client, file_type: str, file_name: str = 'file', duration: typing.Optional[int] = None, ) -> typing.Optional[str]: """Upload a file to Lark and return the file_key, or None if upload fails. Args: file_bytes: Raw file bytes. api_client: Lark API client. file_type: Lark file type, e.g. 'opus', 'mp4', 'pdf', 'doc', etc. file_name: Display name for the file. duration: Duration in milliseconds (for audio files). """ try: if len(file_bytes) > _MAX_LARK_MEDIA_BYTES: raise ValueError('Lark media exceeds the size limit') temp_file_path = await asyncio.to_thread( _write_lark_temp_file, file_bytes, ) try: with open(temp_file_path, 'rb') as upload_file: body_builder = ( CreateFileRequestBody.builder().file_type(file_type).file_name(file_name).file(upload_file) ) if duration is not None: body_builder = body_builder.duration(duration) request = CreateFileRequest.builder().request_body(body_builder.build()).build() response = await api_client.im.v1.file.acreate(request) if not response.success(): print( f'client.im.v1.file.create failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}' ) return None return response.data.file_key finally: os.unlink(temp_file_path) except Exception as e: print(f'Failed to upload file to Lark: {e}') traceback.print_exc() return None @staticmethod async def _get_media_bytes( msg: typing.Union[platform_message.Voice, platform_message.File], ) -> typing.Optional[bytes]: """Get bytes from a Voice or File message (base64, url, or path).""" data = None if msg.base64: try: data = await asyncio.to_thread( _decode_lark_base64_limited, msg.base64, ) except Exception: pass elif msg.url: try: session = httpclient.get_session() async with session.get(msg.url) as resp: if resp.status == 200: data = await httpclient.read_limited( resp, max_bytes=_MAX_LARK_MEDIA_BYTES, ) except Exception: pass elif msg.path: try: data = await asyncio.to_thread( _read_lark_path_limited, str(msg.path), ) except Exception: pass return data @staticmethod async def yiri2target( message_chain: platform_message.MessageChain, api_client: lark_oapi.Client ) -> typing.Tuple[list, list]: """Convert message chain to Lark format. Returns: Tuple of (text_elements, image_keys): - text_elements: List of paragraphs for post message format - media_items: List of dicts with 'msg_type' and 'content' for separate media messages """ message_elements = [] media_items = [] pending_paragraph = [] # Regex pattern to match Markdown image syntax: ![alt](url) markdown_image_pattern = re.compile(r'!\[([^\]]*)\]\(([^)]+)\)') async def process_text_with_images(text: str) -> typing.Tuple[str, list]: """Extract Markdown images from text and return cleaned text + image URLs.""" extracted_urls = [] # Find all Markdown images matches = list(markdown_image_pattern.finditer(text)) if not matches: return text, [] # Extract URLs and remove image syntax from text cleaned_text = text for match in reversed(matches): # Reverse to maintain correct positions url = match.group(2) extracted_urls.insert(0, url) # Insert at beginning since we're going in reverse # Replace image syntax with empty string or a placeholder cleaned_text = cleaned_text[: match.start()] + cleaned_text[match.end() :] # Clean up multiple consecutive newlines that might result from removing images cleaned_text = re.sub(r'\n{3,}', '\n\n', cleaned_text) cleaned_text = cleaned_text.strip() return cleaned_text, extracted_urls for msg in message_chain: if isinstance(msg, platform_message.Plain): # Ensure text is valid UTF-8 try: text = msg.text.encode('utf-8').decode('utf-8') except UnicodeError: try: text = msg.text.encode('latin1').decode('utf-8') except UnicodeError: text = msg.text.encode('utf-8', errors='replace').decode('utf-8') # Check for and extract Markdown images from text cleaned_text, extracted_urls = await process_text_with_images(text) # Split by blank lines to create separate paragraphs for Lark post format. # Lark truncates md elements at the first \n\n, so we must use the # post format's native paragraph structure instead. if cleaned_text: segments = re.split(r'\n\s*\n', cleaned_text) for i, segment in enumerate(segments): segment = segment.strip() if not segment: continue if i > 0 and pending_paragraph: message_elements.append(pending_paragraph) pending_paragraph = [] pending_paragraph.append({'tag': 'md', 'text': segment}) # Process extracted image URLs for url in extracted_urls: temp_image = platform_message.Image(url=url) image_key = await LarkMessageConverter.upload_image_to_lark(temp_image, api_client) if image_key: media_items.append({'msg_type': 'image', 'content': {'image_key': image_key}}) elif isinstance(msg, platform_message.At): pending_paragraph.append({'tag': 'at', 'user_id': msg.target, 'style': []}) elif isinstance(msg, platform_message.AtAll): pending_paragraph.append({'tag': 'at', 'user_id': 'all', 'style': []}) elif isinstance(msg, platform_message.Image): image_key = await LarkMessageConverter.upload_image_to_lark(msg, api_client) if image_key: media_items.append({'msg_type': 'image', 'content': {'image_key': image_key}}) elif isinstance(msg, platform_message.Voice): data = await LarkMessageConverter._get_media_bytes(msg) if data: duration = int(msg.length * 1000) if msg.length else None file_key = await LarkMessageConverter.upload_file_to_lark( data, api_client, file_type='opus', file_name='voice.opus', duration=duration ) if file_key: media_items.append({'msg_type': 'audio', 'content': {'file_key': file_key}}) elif isinstance(msg, platform_message.File): data = await LarkMessageConverter._get_media_bytes(msg) if data: file_name = msg.name or 'file' # Guess file_type from extension ext = os.path.splitext(file_name)[1].lstrip('.').lower() if file_name else '' file_type_map = { 'opus': 'opus', 'mp4': 'mp4', 'pdf': 'pdf', 'doc': 'doc', 'docx': 'doc', 'xls': 'xls', 'xlsx': 'xls', 'ppt': 'ppt', 'pptx': 'ppt', } file_type = file_type_map.get(ext, 'stream') file_key = await LarkMessageConverter.upload_file_to_lark( data, api_client, file_type=file_type, file_name=file_name ) if file_key: media_items.append({'msg_type': 'file', 'content': {'file_key': file_key}}) elif isinstance(msg, platform_message.Forward): for node in msg.node_list: sub_elements, sub_media = await LarkMessageConverter.yiri2target(node.message_chain, api_client) message_elements.extend(sub_elements) media_items.extend(sub_media) if pending_paragraph: message_elements.append(pending_paragraph) return message_elements, media_items @staticmethod async def target2yiri( message: lark_oapi.api.im.v1.model.event_message.EventMessage, api_client: lark_oapi.Client, ) -> platform_message.MessageChain: message_content = json.loads(message.content) lb_msg_list = [] msg_create_time = datetime.datetime.fromtimestamp(int(message.create_time) / 1000) lb_msg_list.append(platform_message.Source(id=message.message_id, time=msg_create_time)) if message.message_type == 'text': element_list = [] def text_element_recur(text_ele: dict) -> list[dict]: if text_ele['text'] == '': return [] at_pattern = re.compile(r'@_user_[\d]+') at_matches = at_pattern.findall(text_ele['text']) name_mapping = {} for mathc in at_matches: for mention in message.mentions: if mention.key == mathc: name_mapping[mathc] = mention.name break if len(name_mapping.keys()) == 0: return [text_ele] # 只处理第一个，剩下的递归处理 text_split = text_ele['text'].split(list(name_mapping.keys())[0]) new_list = [] left_text = text_split[0] right_text = text_split[1] new_list.extend(text_element_recur({'tag': 'text', 'text': left_text, 'style': []})) new_list.append( { 'tag': 'at', 'user_id': list(name_mapping.keys())[0], 'user_name': name_mapping[list(name_mapping.keys())[0]], 'style': [], } ) new_list.extend(text_element_recur({'tag': 'text', 'text': right_text, 'style': []})) return new_list element_list = text_element_recur({'tag': 'text', 'text': message_content['text'], 'style': []}) message_content = {'title': '', 'content': element_list} elif message.message_type == 'post': new_list = [] for ele in message_content['content']: if type(ele) is dict: new_list.append(ele) elif type(ele) is list: new_list.extend(ele) message_content['content'] = new_list elif message.message_type == 'image': message_content['content'] = [{'tag': 'img', 'image_key': message_content['image_key'], 'style': []}] elif message.message_type == 'file': message_content['content'] = [ {'tag': 'file', 'file_key': message_content['file_key'], 'file_name': message_content['file_name']} ] elif message.message_type == 'audio': message_content['content'] = [ { 'tag': 'audio', 'file_key': message_content['file_key'], 'duration': message_content.get('duration', 0), } ] for ele in message_content['content']: if ele['tag'] == 'text': lb_msg_list.append(platform_message.Plain(text=ele['text'])) elif ele['tag'] == 'at': lb_msg_list.append(platform_message.At(target=ele['user_name'])) elif ele['tag'] == 'img': image_key = ele['image_key'] request: GetMessageResourceRequest = ( GetMessageResourceRequest.builder() .message_id(message.message_id) .file_key(image_key) .type('image') .build() ) response: GetMessageResourceResponse = await api_client.im.v1.message_resource.aget(request) if not response.success(): raise Exception( f'client.im.v1.message_resource.get 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)}' ) image_bytes = await asyncio.to_thread( _read_lark_response_file_limited, response, ) image_base64 = (await asyncio.to_thread(base64.b64encode, image_bytes)).decode() image_format = response.raw.headers['content-type'] lb_msg_list.append(platform_message.Image(base64=f'data:{image_format};base64,{image_base64}')) elif ele['tag'] == 'audio': file_key = ele['file_key'] duration = ele['duration'] # Download audio file request: GetMessageResourceRequest = ( GetMessageResourceRequest.builder() .message_id(message.message_id) .file_key(file_key) .type('file') .build() ) try: response: GetMessageResourceResponse = await api_client.im.v1.message_resource.aget(request) if not response.success(): print(f'Failed to download audio: code: {response.code}, msg: {response.msg}') lb_msg_list.append(platform_message.Plain(text='[Audio file download failed]')) return platform_message.MessageChain(lb_msg_list) audio_bytes = await asyncio.to_thread( _read_lark_response_file_limited, response, ) audio_base64 = (await asyncio.to_thread(base64.b64encode, audio_bytes)).decode() # Get content type from response headers content_type = response.raw.headers.get('content-type', 'audio/mpeg') lb_msg_list.append( platform_message.Voice( voice_id=file_key, base64=f'data:{content_type};base64,{audio_base64}', length=(duration // 1000) if duration else None, ) ) except Exception as e: print(f'Error downloading audio: {e}') traceback.print_exc() lb_msg_list.append(platform_message.Plain(text='[Audio file download error]')) elif ele['tag'] == 'file': file_key = ele['file_key'] file_name = ele['file_name'] request: GetMessageResourceRequest = ( GetMessageResourceRequest.builder() .message_id(message.message_id) .file_key(file_key) .type('file') .build() ) response: GetMessageResourceResponse = await api_client.im.v1.message_resource.aget(request) if not response.success(): raise Exception( f'client.im.v1.message_resource.get 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)}' ) file_bytes = await asyncio.to_thread( _read_lark_response_file_limited, response, ) file_base64 = (await asyncio.to_thread(base64.b64encode, file_bytes)).decode() file_format = response.raw.headers['content-type'] file_size = len(file_bytes) lb_msg_list.append( platform_message.File( id=file_key, name=file_name, size=file_size, base64=f'data:{file_format};base64,{file_base64}', ) ) return platform_message.MessageChain(lb_msg_list) class LarkEventConverter(abstract_platform_adapter.AbstractEventConverter): _processed_thread_quote_cache: typing.ClassVar[dict[str, float]] = {} _processed_thread_quote_cache_max_size: typing.ClassVar[int] = 4096 _processed_thread_quote_cache_ttl_seconds: typing.ClassVar[int] = 86400 @classmethod def _prune_processed_thread_quote_cache(cls, now: typing.Optional[float] = None) -> None: if now is None: now = time.time() expire_before = now - cls._processed_thread_quote_cache_ttl_seconds while cls._processed_thread_quote_cache: oldest_key, oldest_ts = next(iter(cls._processed_thread_quote_cache.items())) if oldest_ts >= expire_before: break cls._processed_thread_quote_cache.pop(oldest_key, None) while len(cls._processed_thread_quote_cache) > cls._processed_thread_quote_cache_max_size: oldest_key = next(iter(cls._processed_thread_quote_cache)) cls._processed_thread_quote_cache.pop(oldest_key, None) @classmethod def _mark_thread_quote_processed(cls, thread_id: str) -> None: now = time.time() cls._prune_processed_thread_quote_cache(now) cls._processed_thread_quote_cache[thread_id] = now @classmethod def _extract_quote_message_id(cls, message: EventMessage) -> typing.Optional[str]: """ Extract the message ID to quote from the given message. Rules: - First thread reply in a topic: return parent_id and mark topic as processed - Follow-up thread replies in the same topic: return None - Non-thread message: return parent_id if valid (non-empty, different from message_id) Thread reply state is kept in a bounded TTL cache to avoid unbounded memory growth. """ parent_id = getattr(message, 'parent_id', None) if not parent_id: return None message_id = getattr(message, 'message_id', None) if parent_id == message_id: return None thread_id = getattr(message, 'thread_id', None) if thread_id: cls._prune_processed_thread_quote_cache() if thread_id in cls._processed_thread_quote_cache: return None cls._mark_thread_quote_processed(thread_id) return parent_id @staticmethod def _build_event_message_from_message_item(message_item: Message) -> typing.Optional[EventMessage]: """ Build EventMessage from SDK typed Message item. Returns None if body or content is missing. """ body = getattr(message_item, 'body', None) if not body: return None content = getattr(body, 'content', None) if not content: return None event_data = { 'message_id': message_item.message_id, 'message_type': message_item.msg_type, 'content': content, 'create_time': message_item.create_time, 'mentions': getattr(message_item, 'mentions', []) or [], } # Preserve thread-related fields if hasattr(message_item, 'parent_id') and message_item.parent_id: event_data['parent_id'] = message_item.parent_id if hasattr(message_item, 'root_id') and message_item.root_id: event_data['root_id'] = message_item.root_id if hasattr(message_item, 'thread_id') and message_item.thread_id: event_data['thread_id'] = message_item.thread_id if hasattr(message_item, 'chat_id') and message_item.chat_id: event_data['chat_id'] = message_item.chat_id return EventMessage(event_data) @staticmethod async def _fetch_quoted_message( quote_message_id: str, api_client: lark_oapi.Client, ) -> typing.Optional[platform_message.MessageChain]: """ Fetch the quoted message and convert to MessageChain. Returns None if: - API call fails - Response items is empty - Message item normalization fails """ request = GetMessageRequest.builder().message_id(quote_message_id).build() response = await api_client.im.v1.message.aget(request) if not response.success(): return None items = getattr(response.data, 'items', None) if not items: return None message_item = items[0] event_message = LarkEventConverter._build_event_message_from_message_item(message_item) if event_message is None: return None quote_chain = await LarkMessageConverter.target2yiri(event_message, api_client) return quote_chain @staticmethod async def yiri2target( event: platform_events.MessageEvent, ) -> lark_oapi.im.v1.P2ImMessageReceiveV1: pass @staticmethod async def target2yiri( event: lark_oapi.im.v1.P2ImMessageReceiveV1, api_client: lark_oapi.Client ) -> platform_events.Event: message_chain = await LarkMessageConverter.target2yiri(event.event.message, api_client) # Check for quote/reply message # Extract files/images/voice from quote and add them as top-level components # so that plugins like FileReader can process them the same way as direct messages quote_message_id = LarkEventConverter._extract_quote_message_id(event.event.message) if quote_message_id: quote_chain = await LarkEventConverter._fetch_quoted_message(quote_message_id, api_client) if quote_chain: # Filter out Source component from quoted chain, keep only content quote_components = [comp for comp in quote_chain if not isinstance(comp, platform_message.Source)] # Add quoted content as top-level components instead of wrapping in Quote for comp in quote_components: if isinstance(comp, platform_message.File): # Add file as top-level component (same as direct message) message_chain.append(comp) elif isinstance(comp, platform_message.Image): # Add image as top-level component message_chain.append(comp) elif isinstance(comp, platform_message.Voice): # Add voice as top-level component message_chain.append(comp) elif isinstance(comp, platform_message.Plain): # Add text with context prefix message_chain.append(platform_message.Plain(text=f'[引用消息] {comp.text}')) if event.event.message.chat_type == 'p2p': return platform_events.FriendMessage( sender=platform_entities.Friend( id=event.event.sender.sender_id.open_id, nickname=event.event.sender.sender_id.union_id, remark='', ), message_chain=message_chain, time=event.event.message.create_time, source_platform_object=event, ) elif event.event.message.chat_type == 'group': return platform_events.GroupMessage( sender=platform_entities.GroupMember( id=event.event.sender.sender_id.open_id, member_name=event.event.sender.sender_id.union_id, permission=platform_entities.Permission.Member, group=platform_entities.Group( id=event.event.message.chat_id, name='', permission=platform_entities.Permission.Member, ), special_title='', ), message_chain=message_chain, time=event.event.message.create_time, source_platform_object=event, ) CARD_ID_CACHE_SIZE = 500 CARD_ID_CACHE_MAX_LIFETIME = 20 * 60 # 20分钟 class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): bot: lark_oapi.ws.Client = pydantic.Field(exclude=True) api_client: lark_oapi.Client = pydantic.Field(exclude=True) ap: typing.Any = pydantic.Field(exclude=True, default=None) bot_account_id: str # 用于在流水线中识别at是否是本bot，直接以bot_name作为标识 lark_tenant_key: str = pydantic.Field(exclude=True, default='') # 飞书企业key message_converter: LarkMessageConverter = LarkMessageConverter() event_converter: LarkEventConverter = LarkEventConverter() cipher: AESCipher listeners: typing.Dict[ typing.Type[platform_events.Event], typing.Callable[[platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None], ] quart_app: quart.Quart = pydantic.Field(exclude=True) card_id_dict: dict[str, str] # 消息id到卡片id的映射，便于创建卡片后的发送消息到指定卡片 # Monitoring message ID mapping for feedback correlation # Temp: user Lark message ID → monitoring_message_id (populated by on_monitoring_message_created, consumed by create_message_card) pending_monitoring_msg: dict[str, str] # Final: reply Lark message ID → (monitoring_message_id, timestamp) (used by feedback callbacks) reply_to_monitoring_msg: dict[str, tuple[str, float]] reply_message_card_ids: dict[str, str] card_sequence_dict: dict[str, int] # card_id → set of source message ids registered against it (for cleanup) card_id_to_source_ids: dict[str, set[str]] # card_id → current streaming_txt content cache (needed for full aupdate during resume transition) card_streaming_text: dict[str, str] # card_id → pre-pause streaming_txt text (captured when resume first chunk arrives) card_pre_pause_text: dict[str, str] # card_id → form_content captured when the form is first shown (for resume notice) card_form_content: dict[str, str] # card_id → input_defs / inputs captured for the selected-action notice card_form_input_defs: dict[str, list[dict]] card_form_inputs: dict[str, dict] card_last_accessed: dict[str, float] card_cleanup_at: float # set of card_ids that have already transitioned from "buttons visible" to "resume layout" card_resume_transitioned: set[str] inbound_event_tasks: set[asyncio.Task] threadsafe_event_futures: set[typing.Any] threadsafe_event_lock: typing.Any = pydantic.Field(exclude=True) _MONITORING_MAPPING_TTL = 600 # 10 minutes _MAX_INBOUND_EVENTS = 100 _MAX_TENANT_ACCESS_TOKENS = 1024 seq: int # 用于在发送卡片消息中识别消息顺序，直接以seq作为标识 bot_uuid: str = None # 机器人UUID app_ticket: str = None # 商店应用用到 app_access_token: str = None # 商店应用用到 app_access_token_expire_at: int = None tenant_access_tokens: dict[str, dict[str, str]] = pydantic.Field( default_factory=dict, exclude=True, ) def __init__(self, config: dict, logger: abstract_platform_logger.AbstractEventLogger, **kwargs): quart_app = quart.Quart(__name__) async def on_message(event: lark_oapi.im.v1.P2ImMessageReceiveV1): lb_event = await self.event_converter.target2yiri(event, self.api_client) await self.listeners[type(lb_event)](lb_event, self) def sync_on_message(event: lark_oapi.im.v1.P2ImMessageReceiveV1): self._schedule_inbound_event(on_message(event)) def schedule_on_app_loop(coro): """Run a coroutine on the application event loop from sync callbacks.""" return self._schedule_threadsafe_event(coro) def sync_on_card_action(event): try: action_value_raw = getattr(getattr(event.event, 'action', None), 'value', {}) # Parse JSON string values (from form action buttons) action_value_obj = _lark_mapping_from_value(action_value_raw) action_value = action_value_obj.get('feedback', '') if isinstance(action_value_obj, dict) else '' # Handle Dify form action button clicks if isinstance(action_value_obj, dict) and action_value_obj.get('form_action'): form_token = action_value_obj.get('form_token', '') workflow_run_id = action_value_obj.get('workflow_run_id', '') action_id = action_value_obj.get('action_id', '') session_key = action_value_obj.get('session_key', '') action = getattr(event.event, 'action', None) form_inputs = _lark_extract_action_form_inputs(action, action_value_obj) if session_key.startswith('group_') or session_key.startswith('g:'): launcher_type = provider_session.LauncherTypes.GROUP launcher_id = ( session_key.split(':', 1)[1] if session_key.startswith('g:') else session_key[len('group_') :] ) else: launcher_type = provider_session.LauncherTypes.PERSON launcher_id = ( session_key.split(':', 1)[1] if session_key.startswith('p:') else session_key[len('person_') :] ) # Find the bot entity to get bot_uuid and pipeline_uuid bot_uuid = '' pipeline_uuid = action_value_obj.get('pipeline_uuid') or None for bot in self.ap.platform_mgr.bots: if bot.adapter is self: bot_uuid = bot.bot_entity.uuid pipeline_uuid = pipeline_uuid or bot.bot_entity.use_pipeline_uuid break form_action_data = { 'form_token': form_token, 'workflow_run_id': workflow_run_id, 'action_id': action_id, 'user': f'{launcher_type.value}_{launcher_id}', 'inputs': form_inputs, } if action_value_obj.get('_input_progress'): form_action_data['_input_progress'] = True context = getattr(event.event, 'context', None) open_message_id = getattr(context, 'open_message_id', None) if open_message_id and form_inputs: card_id = self.reply_message_card_ids.get(str(open_message_id)) else: card_id = None if not card_id: card_id = str(action_value_obj.get('card_id') or '') if card_id and form_inputs: cached_inputs = dict(self.card_form_inputs.get(card_id) or {}) cached_inputs.update(form_inputs) self.card_form_inputs[card_id] = cached_inputs if self.ap is not None: self.ap.logger.info( f'Lark form action inputs cached: card_id={card_id} ' f'open_message_id={open_message_id} keys={list(form_inputs.keys())}' ) source_time = datetime.datetime.now() event_time = source_time.timestamp() action_text = action_value_obj.get('action_id', 'confirm') message_chain = platform_message.MessageChain( [platform_message.Plain(text=f'[Form Action: {action_text}]')] ) if open_message_id: message_chain.insert( 0, platform_message.Source( id=open_message_id, time=source_time, ), ) operator = getattr(event.event, 'operator', None) user_id = ( getattr(operator, 'open_id', None) or getattr(operator, 'user_id', None) or str(launcher_id) ) if launcher_type == provider_session.LauncherTypes.GROUP: synthetic_event = platform_events.GroupMessage( sender=platform_entities.GroupMember( id=user_id, member_name='', permission=platform_entities.Permission.Member, group=platform_entities.Group( id=launcher_id, name='', permission=platform_entities.Permission.Member, ), ), message_chain=message_chain, time=event_time, source_platform_object=event, ) else: synthetic_event = platform_events.FriendMessage( sender=platform_entities.Friend( id=user_id, nickname='', remark='', ), message_chain=message_chain, time=event_time, source_platform_object=event, ) async def add_form_action_query(): await self.ap.query_pool.add_query( bot_uuid=bot_uuid, launcher_type=launcher_type, launcher_id=launcher_id, sender_id=user_id, message_event=synthetic_event, message_chain=message_chain, adapter=self, pipeline_uuid=pipeline_uuid, variables={ '_dify_form_action': form_action_data, '_routed_by_rule': True, }, ) schedule_on_app_loop(add_form_action_query()) from lark_oapi.event.callback.model.p2_card_action_trigger import P2CardActionTriggerResponse return P2CardActionTriggerResponse({'toast': {'type': 'success', 'content': '操作成功'}}) if action_value == '有帮助': feedback_type = 1 elif action_value == '无帮助': feedback_type = 2 else: from lark_oapi.event.callback.model.p2_card_action_trigger import P2CardActionTriggerResponse return P2CardActionTriggerResponse({'toast': {'type': 'success', 'content': '操作成功'}}) operator = getattr(event.event, 'operator', None) context = getattr(event.event, 'context', None) user_id = getattr(operator, 'open_id', None) or getattr(operator, 'user_id', None) open_chat_id = getattr(context, 'open_chat_id', None) open_message_id = getattr(context, 'open_message_id', None) if open_chat_id: session_id = f'group_{open_chat_id}' elif user_id: session_id = f'person_{user_id}' else: session_id = None # Resolve monitoring message ID from reply message mapping monitoring_msg_id = None if open_message_id and open_message_id in self.reply_to_monitoring_msg: monitoring_msg_id = self.reply_to_monitoring_msg[open_message_id][0] feedback_event = platform_events.FeedbackEvent( feedback_id=getattr(event.header, 'event_id', str(uuid.uuid4())), feedback_type=feedback_type, feedback_content=action_value, user_id=user_id, session_id=session_id, message_id=open_message_id, stream_id=monitoring_msg_id, source_platform_object=event, ) if platform_events.FeedbackEvent in self.listeners: schedule_on_app_loop(self.listeners[platform_events.FeedbackEvent](feedback_event, self)) from lark_oapi.event.callback.model.p2_card_action_trigger import P2CardActionTriggerResponse return P2CardActionTriggerResponse({'toast': {'type': 'success', 'content': '感谢您的反馈'}}) except Exception: traceback.print_exc() schedule_on_app_loop(self.logger.error(f'Error in lark card action callback: {traceback.format_exc()}')) from lark_oapi.event.callback.model.p2_card_action_trigger import P2CardActionTriggerResponse return P2CardActionTriggerResponse({'toast': {'type': 'error', 'content': '反馈处理失败'}}) event_handler = ( lark_oapi.EventDispatcherHandler.builder('', '') .register_p2_im_message_receive_v1(sync_on_message) .register_p2_card_action_trigger(sync_on_card_action) .build() ) bot_account_id = config['bot_name'] domain = self._resolve_domain(config) bot = NonBlockingLarkWSClient( config['app_id'], config['app_secret'], event_handler=event_handler, domain=domain ) api_client = self.build_api_client(config) cipher = AESCipher(config.get('encrypt-key', '')) self.request_app_ticket(api_client, config) super().__init__( config=config, logger=logger, lark_tenant_key=config.get('lark_tenant_key', ''), card_id_dict={}, pending_monitoring_msg={}, reply_to_monitoring_msg={}, reply_message_card_ids={}, card_sequence_dict={}, card_id_to_source_ids={}, card_streaming_text={}, card_pre_pause_text={}, card_form_content={}, card_form_input_defs={}, card_form_inputs={}, card_last_accessed={}, card_cleanup_at=0.0, card_resume_transitioned=set(), inbound_event_tasks=set(), threadsafe_event_futures=set(), threadsafe_event_lock=threading.Lock(), tenant_access_tokens={}, seq=1, listeners={}, quart_app=quart_app, bot=bot, api_client=api_client, bot_account_id=bot_account_id, cipher=cipher, **kwargs, ) def _schedule_inbound_event(self, coro) -> None: for task in tuple(self.inbound_event_tasks): if task.done(): self.inbound_event_tasks.discard(task) if len(self.inbound_event_tasks) >= self._MAX_INBOUND_EVENTS: coro.close() return task = asyncio.create_task(coro) self.inbound_event_tasks.add(task) def done(done_task: asyncio.Task) -> None: self.inbound_event_tasks.discard(done_task) if not done_task.cancelled(): done_task.exception() task.add_done_callback(done) def _schedule_threadsafe_event(self, coro): """Submit one bounded callback from the Lark SDK's sync boundary.""" with self.threadsafe_event_lock: for future in tuple(self.threadsafe_event_futures): if future.done(): self.threadsafe_event_futures.discard(future) if len(self.threadsafe_event_futures) >= self._MAX_INBOUND_EVENTS: coro.close() return None future = asyncio.run_coroutine_threadsafe(coro, self.ap.event_loop) self.threadsafe_event_futures.add(future) def done(done_future) -> None: with self.threadsafe_event_lock: self.threadsafe_event_futures.discard(done_future) if not done_future.cancelled(): done_future.exception() future.add_done_callback(done) return future def request_app_ticket(self, api_client, config): app_id = config['app_id'] app_secret = config['app_secret'] print(f'Requesting app ticket for app_id: {app_id[:3]}***{app_id[-3:]}') if 'isv' == config.get('app_type', 'self'): request: ResendAppTicketRequest = ( ResendAppTicketRequest.builder() .request_body(ResendAppTicketRequestBody.builder().app_id(app_id).app_secret(app_secret).build()) .build() ) response: ResendAppTicketResponse = api_client.auth.v3.app_ticket.resend(request) if not response.success(): raise Exception( f'client.auth.v3.auth.app_ticket_resend 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)}' ) def request_app_access_token(self): app_id = self.config['app_id'] app_secret = self.config['app_secret'] if 'isv' == self.config.get('app_type', 'self'): request: CreateAppAccessTokenRequest = ( CreateAppAccessTokenRequest.builder() .request_body( CreateAppAccessTokenRequestBody.builder() .app_id(app_id) .app_secret(app_secret) .app_ticket(self.app_ticket) .build() ) .build() ) response: CreateAppAccessTokenResponse = self.api_client.auth.v3.app_access_token.create(request) if not response.success(): raise Exception( f'client.auth.v3.auth.app_access_token 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)}' ) content = json.loads(response.raw.content) self.app_access_token = content['app_access_token'] self.app_access_token_expire_at = int(time.time()) + content['expire'] - 300 def get_app_access_token(self): if 'isv' != self.config.get('app_type', 'self'): return None if ( self.app_access_token is None or self.app_access_token_expire_at is None or int(time.time()) >= self.app_access_token_expire_at ): self.request_app_access_token() return self.app_access_token def request_tenant_access_token(self, tenant_key: str): app_access_token = self.get_app_access_token() if 'isv' == self.config.get('app_type', 'self'): request: CreateTenantAccessTokenRequest = ( CreateTenantAccessTokenRequest.builder() .request_body( CreateTenantAccessTokenRequestBody.builder() .app_access_token(app_access_token) .tenant_key(tenant_key) .build() ) .build() ) response: CreateTenantAccessTokenResponse = self.api_client.auth.v3.tenant_access_token.create(request) if not response.success(): raise Exception( f'client.auth.v3.auth.tenant_access_token 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)}' ) content = json.loads(response.raw.content) tenant_access_token = content['tenant_access_token'] expire = content['expire'] self.tenant_access_tokens[tenant_key] = { 'token': tenant_access_token, 'expire_at': int(time.time()) + expire - 300, } now = int(time.time()) for cached_key, cached_token in tuple(self.tenant_access_tokens.items()): if int(cached_token.get('expire_at', 0)) <= now: self.tenant_access_tokens.pop(cached_key, None) while len(self.tenant_access_tokens) > self._MAX_TENANT_ACCESS_TOKENS: self.tenant_access_tokens.pop(next(iter(self.tenant_access_tokens)), None) def get_tenant_access_token(self, tenant_key: str): if tenant_key is None or 'isv' != self.config.get('app_type', 'self'): return None tenant_access_token = self.tenant_access_tokens.get(tenant_key) if tenant_access_token is None or int(time.time()) >= tenant_access_token['expire_at']: self.request_tenant_access_token(tenant_key) return self.tenant_access_tokens.get(tenant_key)['token'] if self.tenant_access_tokens.get(tenant_key) else None def get_launcher_id(self, event: platform_events.MessageEvent) -> str | None: """ Get topic-scoped launcher_id for thread-aware session isolation. For group thread messages, returns "{group_id}_{thread_id}" to ensure conversation context stays stable per topic. Returns None for non-thread messages or P2P messages. """ source_event = getattr(event.source_platform_object, 'event', None) if not source_event: return None message = getattr(source_event, 'message', None) if not message: return None thread_id = getattr(message, 'thread_id', None) if not thread_id: return None if isinstance(event, platform_events.GroupMessage): return f'{event.group.id}_{thread_id}' return None @staticmethod def _resolve_domain(config) -> str: domain = config.get('domain', lark_oapi.FEISHU_DOMAIN) if domain == 'custom': domain = config.get('custom_domain', '') if not domain: raise ValueError('Custom domain is required when domain is set to "custom"') return domain.rstrip('/') def build_api_client(self, config): app_id = config['app_id'] app_secret = config['app_secret'] domain = self._resolve_domain(config) api_client = lark_oapi.Client.builder().app_id(app_id).app_secret(app_secret).domain(domain).build() if 'isv' == config.get('app_type', 'self'): api_client = ( lark_oapi.Client.builder() .app_id(app_id) .app_secret(app_secret) .app_type(lark_oapi.AppType.ISV) .domain(domain) .build() ) return api_client @staticmethod def _has_markdown_table(text_elements: list) -> bool: """Check if text elements contain markdown table syntax (|...|). A markdown table requires: - At least one line with pipe characters (|) - A separator line with pipes and dashes (|---|) """ # Regex to detect markdown table: lines with pipes and separator lines with dashes table_pattern = re.compile(r'^\s*\|.+\|\s*$', re.MULTILINE) separator_pattern = re.compile(r'^\s*\|\s*-+(\s*\|\s*-+)*\s*\|\s*$', re.MULTILINE) for paragraph in text_elements: for ele in paragraph: if ele.get('tag') == 'md': text = ele.get('text', '') # Quick check: if has pipes and separator line, it's likely a table if '|' in text and table_pattern.search(text) and separator_pattern.search(text): return True return False async def send_message(self, target_type: str, target_id: str, message: platform_message.MessageChain): text_elements, media_items = await self.message_converter.yiri2target(message, self.api_client) # Map standard target_type to Feishu receive_id_type if target_type == 'person': receive_id_type = 'open_id' elif target_type == 'group': receive_id_type = 'chat_id' else: receive_id_type = target_type # Send text message if there are text elements if text_elements: # Use 'post' format if: has @mentions OR has markdown tables has_at = any(ele['tag'] == 'at' for paragraph in text_elements for ele in paragraph) has_table = self._has_markdown_table(text_elements) needs_post = has_at or has_table if needs_post: msg_type = 'post' final_content = json.dumps( { 'zh_Hans': { 'title': '', 'content': text_elements, }, } ) else: msg_type = 'text' parts = [] for paragraph in text_elements: para_text = ''.join(ele.get('text', '') for ele in paragraph) if para_text: parts.append(para_text) final_content = json.dumps({'text': '\n\n'.join(parts)}) request: CreateMessageRequest = ( CreateMessageRequest.builder() .receive_id_type(receive_id_type) .request_body( CreateMessageRequestBody.builder() .receive_id(target_id) .content(final_content) .msg_type(msg_type) .uuid(str(uuid.uuid4())) .build() ) .build() ) app_access_token = self.get_app_access_token() req_opt: RequestOption = ( RequestOption.builder().app_ticket(self.app_ticket).app_access_token(app_access_token).build() ) response: CreateMessageResponse = self.api_client.im.v1.message.create(request, req_opt) if not response.success(): raise Exception( 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)}' ) # Send media messages separately (image, audio, file, etc.) for media in media_items: request: CreateMessageRequest = ( CreateMessageRequest.builder() .receive_id_type(receive_id_type) .request_body( CreateMessageRequestBody.builder() .receive_id(target_id) .content(json.dumps(media['content'])) .msg_type(media['msg_type']) .uuid(str(uuid.uuid4())) .build() ) .build() ) app_access_token = self.get_app_access_token() req_opt: RequestOption = ( RequestOption.builder().app_ticket(self.app_ticket).app_access_token(app_access_token).build() ) response: CreateMessageResponse = self.api_client.im.v1.message.create(request, req_opt) if not response.success(): raise Exception( f'client.im.v1.message.create ({media["msg_type"]}) 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)}' ) async def is_stream_output_supported(self) -> bool: is_stream = False if self.config.get('enable-stream-reply', None): is_stream = True return is_stream async def on_monitoring_message_created(self, query, monitoring_message_id: str): """Called by pipeline after monitoring message is created, to map user message ID to monitoring message ID.""" try: user_msg_id = query.message_event.message_chain.message_id if user_msg_id: self.pending_monitoring_msg[user_msg_id] = monitoring_message_id while len(self.pending_monitoring_msg) > CARD_ID_CACHE_SIZE: self.pending_monitoring_msg.pop(next(iter(self.pending_monitoring_msg)), None) except Exception as e: await self.logger.debug(f'Failed to map message to monitoring message: {e}') def _cleanup_monitoring_mapping(self): """Remove entries older than TTL from the reply-to-monitoring mapping.""" now = time.time() expired = [k for k, (_, ts) in self.reply_to_monitoring_msg.items() if now - ts > self._MONITORING_MAPPING_TTL] for k in expired: del self.reply_to_monitoring_msg[k] def _next_card_sequence(self, card_id: str, suggested: int = 1) -> int: """Return the next strictly increasing sequence for a card update.""" self._touch_card(card_id) current = self.card_sequence_dict.get(card_id, 0) next_seq = max(current + 1, suggested) self.card_sequence_dict[card_id] = next_seq return next_seq def _register_card_for_source(self, card_id: str, *source_ids: str) -> None: """Register a card_id under one or more source message ids.""" self._touch_card(card_id) bucket = self.card_id_to_source_ids.setdefault(card_id, set()) for sid in source_ids: if not sid: continue self.reply_message_card_ids[sid] = card_id bucket.add(sid) def _drop_card_state(self, card_id: str) -> None: """Pop all per-card state for the given card_id.""" if not card_id: return for sid in self.card_id_to_source_ids.pop(card_id, set()): self.reply_message_card_ids.pop(sid, None) self.card_sequence_dict.pop(card_id, None) self.card_streaming_text.pop(card_id, None) self.card_pre_pause_text.pop(card_id, None) self.card_form_content.pop(card_id, None) self.card_form_input_defs.pop(card_id, None) self.card_form_inputs.pop(card_id, None) self.card_last_accessed.pop(card_id, None) self.card_resume_transitioned.discard(card_id) def _touch_card(self, card_id: str) -> None: now = time.monotonic() if now - self.card_cleanup_at >= 60 or len(self.card_last_accessed) >= CARD_ID_CACHE_SIZE: self.card_cleanup_at = now for stale_card_id, last_accessed in tuple(self.card_last_accessed.items()): if now - last_accessed >= CARD_ID_CACHE_MAX_LIFETIME: self._drop_card_state(stale_card_id) while len(self.card_last_accessed) >= CARD_ID_CACHE_SIZE: oldest_card_id = min( self.card_last_accessed, key=self.card_last_accessed.__getitem__, ) self._drop_card_state(oldest_card_id) self.card_last_accessed[card_id] = now async def create_card_id(self, message_id): try: # self.logger.debug('飞书支持stream输出,创建卡片......') card_data = { 'schema': '2.0', 'config': { 'update_multi': True, 'streaming_mode': True, 'streaming_config': { 'print_step': {'default': 1}, 'print_frequency_ms': {'default': 70}, 'print_strategy': 'fast', }, }, 'body': { 'direction': 'vertical', 'padding': '12px 12px 12px 12px', 'elements': [ { 'tag': 'div', 'text': { 'tag': 'plain_text', 'content': 'LangBot', 'text_size': 'normal', 'text_align': 'left', 'text_color': 'default', }, 'icon': { 'tag': 'custom_icon', 'img_key': 'img_v3_02p3_05c65d5d-9bad-440a-a2fb-c89571bfd5bg', }, }, { 'tag': 'markdown', 'content': '', 'text_align': 'left', 'text_size': 'normal', 'margin': '0px 0px 0px 0px', 'element_id': 'streaming_txt', }, { 'tag': 'markdown', 'content': '', 'text_align': 'left', 'text_size': 'normal', 'margin': '0px 0px 0px 0px', }, { 'tag': 'column_set', 'horizontal_spacing': '8px', 'horizontal_align': 'left', 'columns': [ { 'tag': 'column', 'width': 'weighted', 'elements': [ { 'tag': 'markdown', 'content': '', 'text_align': 'left', 'text_size': 'normal', 'margin': '0px 0px 0px 0px', }, { 'tag': 'markdown', 'content': '', 'text_align': 'left', 'text_size': 'normal', 'margin': '0px 0px 0px 0px', }, { 'tag': 'markdown', 'content': '', 'text_align': 'left', 'text_size': 'normal', 'margin': '0px 0px 0px 0px', }, ], 'padding': '0px 0px 0px 0px', 'direction': 'vertical', 'horizontal_spacing': '8px', 'vertical_spacing': '2px', 'horizontal_align': 'left', 'vertical_align': 'top', 'margin': '0px 0px 0px 0px', 'weight': 1, } ], 'margin': '0px 0px 0px 0px', }, {'tag': 'hr', 'margin': '0px 0px 0px 0px'}, { 'tag': 'column_set', 'horizontal_spacing': '12px', 'horizontal_align': 'right', 'columns': [ { 'tag': 'column', 'width': 'weighted', 'elements': [ { 'tag': 'markdown', 'content': '以上内容由 AI 生成，仅供参考。更多详细、准确信息可点击引用链接查看', 'text_align': 'left', 'text_size': 'notation', 'margin': '4px 0px 0px 0px', 'icon': { 'tag': 'standard_icon', 'token': 'robot_outlined', 'color': 'grey', }, } ], 'padding': '0px 0px 0px 0px', 'direction': 'vertical', 'horizontal_spacing': '8px', 'vertical_spacing': '8px', 'horizontal_align': 'left', 'vertical_align': 'top', 'margin': '0px 0px 0px 0px', 'weight': 1, }, { 'tag': 'column', 'width': '20px', 'elements': [ { 'tag': 'button', 'text': {'tag': 'plain_text', 'content': ''}, 'type': 'text', 'width': 'fill', 'size': 'medium', 'icon': {'tag': 'standard_icon', 'token': 'thumbsup_outlined'}, 'hover_tips': {'tag': 'plain_text', 'content': '有帮助'}, 'behaviors': [{'type': 'callback', 'value': {'feedback': '有帮助'}}], 'margin': '0px 0px 0px 0px', } ], 'padding': '0px 0px 0px 0px', 'direction': 'vertical', 'horizontal_spacing': '8px', 'vertical_spacing': '8px', 'horizontal_align': 'left', 'vertical_align': 'top', 'margin': '0px 0px 0px 0px', }, { 'tag': 'column', 'width': '30px', 'elements': [ { 'tag': 'button', 'text': {'tag': 'plain_text', 'content': ''}, 'type': 'text', 'width': 'default', 'size': 'medium', 'icon': {'tag': 'standard_icon', 'token': 'thumbdown_outlined'}, 'hover_tips': {'tag': 'plain_text', 'content': '无帮助'}, 'behaviors': [{'type': 'callback', 'value': {'feedback': '无帮助'}}], 'margin': '0px 0px 0px 0px', } ], 'padding': '0px 0px 0px 0px', 'vertical_spacing': '8px', 'horizontal_align': 'left', 'vertical_align': 'top', 'margin': '0px 0px 0px 0px', }, ], 'margin': '0px 0px 4px 0px', }, ], }, } # delay / fast 创建卡片模板，delay 延迟打印，fast 实时打印，可以自定义更好看的消息模板 request: CreateCardRequest = ( CreateCardRequest.builder() .request_body(CreateCardRequestBody.builder().type('card_json').data(json.dumps(card_data)).build()) .build() ) # 发起请求 response: CreateCardResponse = self.api_client.cardkit.v1.card.create(request) # 处理失败返回 if not response.success(): raise Exception( f'client.cardkit.v1.card.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)}' ) self.card_id_dict[message_id] = response.data.card_id card_id = response.data.card_id self._touch_card(card_id) self.card_sequence_dict[card_id] = 0 return card_id except Exception as e: raise e async def create_message_card(self, message_id, event) -> str: """ 创建卡片消息。 使用卡片消息是因为普通消息更新次数有限制，而大模型流式返回结果可能很多而超过限制，而飞书卡片没有这个限制（api免费次数有限） """ # message_id = event.message_chain.message_id source_message_id = str(event.message_chain.message_id) existing_card_id = self.reply_message_card_ids.get(source_message_id) if existing_card_id: self.card_id_dict[message_id] = existing_card_id return True card_id = await self.create_card_id(message_id) content = { 'type': 'card', 'data': {'card_id': card_id, 'template_variable': {'content': 'Thinking...'}}, } # 当收到消息时发送消息模板，可添加模板变量，详情查看飞书中接口文档 request: ReplyMessageRequest = ( ReplyMessageRequest.builder() .message_id(event.message_chain.message_id) .request_body( ReplyMessageRequestBody.builder().content(json.dumps(content)).msg_type('interactive').build() ) .build() ) tenant_key = event.source_platform_object.header.tenant_key if event.source_platform_object else None app_access_token = self.get_app_access_token() tenant_access_token = self.get_tenant_access_token(tenant_key) req_opt: RequestOption = ( RequestOption.builder() .app_ticket(self.app_ticket) .tenant_key(tenant_key) .app_access_token(app_access_token) .tenant_access_token(tenant_access_token) .build() ) # 发起请求 response: ReplyMessageResponse = await self.api_client.im.v1.message.areply(request, req_opt) # 处理失败返回 if not response.success(): raise Exception( f'client.im.v1.message.reply 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)}' ) # Transfer monitoring message mapping: user msg ID → reply msg ID try: user_msg_id = event.message_chain.message_id reply_msg_id = getattr(response.data, 'message_id', None) monitoring_msg_id = self.pending_monitoring_msg.pop(user_msg_id, None) # Register the card under both the user-incoming msg id (so a # second reply_message_first_chunk for the same user message # reuses this card) AND the bot-reply msg id (so a synthetic # event from a form-button callback — whose Source.id equals # the bot's card message id — hits the same card and renders # the resume content into it). if reply_msg_id: self._register_card_for_source(card_id, str(user_msg_id), str(reply_msg_id)) else: self._register_card_for_source(card_id, str(user_msg_id)) if reply_msg_id and monitoring_msg_id: self.reply_to_monitoring_msg[reply_msg_id] = (monitoring_msg_id, time.time()) self._cleanup_monitoring_mapping() except Exception as e: await self.logger.debug(f'Failed to transfer monitoring mapping in create_message_card: {e}') return True async def _open_new_form_card( self, message_id: str, message_source: platform_events.MessageEvent, form_data: dict, ) -> str | None: """Spawn a fresh card to host a re-paused human-input prompt. Creates a new card_id (rebinding ``self.card_id_dict[message_id]``), replies it to the current incoming message so it appears as the next step in the chat, registers the new reply_msg_id so subsequent button callbacks resolve back to it, and renders the prompt + buttons on it. Returns the new card_id, or ``None`` if creation failed (caller is responsible for falling back to in-place update so the workflow remains continuable). """ source_message_id = getattr(message_source.message_chain, 'message_id', None) if not source_message_id: await self.logger.error('Cannot open new form card: source message_id missing') return None try: new_card_id = await self.create_card_id(message_id) except Exception: await self.logger.error(f'Failed to create new form card: {traceback.format_exc()}') return None tenant_key = ( message_source.source_platform_object.header.tenant_key if message_source.source_platform_object else None ) app_access_token = self.get_app_access_token() tenant_access_token = self.get_tenant_access_token(tenant_key) req_opt: RequestOption = ( RequestOption.builder() .app_ticket(self.app_ticket) .tenant_key(tenant_key) .app_access_token(app_access_token) .tenant_access_token(tenant_access_token) .build() ) content = { 'type': 'card', 'data': {'card_id': new_card_id, 'template_variable': {'content': ''}}, } request: ReplyMessageRequest = ( ReplyMessageRequest.builder() .message_id(str(source_message_id)) .request_body( ReplyMessageRequestBody.builder() .content(json.dumps(content)) .msg_type('interactive') .uuid(str(uuid.uuid4())) .build() ) .build() ) try: response: ReplyMessageResponse = await self.api_client.im.v1.message.areply(request, req_opt) except Exception: await self.logger.error(f'Failed to send new form card: {traceback.format_exc()}') return None if not response.success(): await self.logger.error( f'Failed to send new form card: code={response.code}, msg={response.msg}, ' f'log_id={response.get_log_id()}' ) return None reply_msg_id = getattr(response.data, 'message_id', None) if reply_msg_id: self._register_card_for_source(new_card_id, str(source_message_id), str(reply_msg_id)) sequence = self._next_card_sequence(new_card_id, 1) await self._update_card_layout( card_id=new_card_id, message_source=message_source, text_message='', sequence=sequence, form_data=form_data, show_form_prompt=True, ) return new_card_id async def reply_message( self, message_source: platform_events.MessageEvent, message: platform_message.MessageChain, quote_origin: bool = False, ): # 不再需要了，因为message_id已经被包含到message_chain中 # lark_event = await self.event_converter.yiri2target(message_source) text_elements, media_items = await self.message_converter.yiri2target(message, self.api_client) # Send text message if there are text elements if text_elements: # Use 'post' format if: has @mentions OR has markdown tables has_at = any(ele['tag'] == 'at' for paragraph in text_elements for ele in paragraph) has_table = self._has_markdown_table(text_elements) needs_post = has_at or has_table if needs_post: msg_type = 'post' final_content = json.dumps( { 'zh_Hans': { 'title': '', 'content': text_elements, }, } ) else: msg_type = 'text' parts = [] for paragraph in text_elements: para_text = ''.join(ele.get('text', '') for ele in paragraph) if para_text: parts.append(para_text) final_content = json.dumps({'text': '\n\n'.join(parts)}) request: ReplyMessageRequest = ( ReplyMessageRequest.builder() .message_id(message_source.message_chain.message_id) .request_body( ReplyMessageRequestBody.builder() .content(final_content) .msg_type(msg_type) .reply_in_thread(False) .uuid(str(uuid.uuid4())) .build() ) .build() ) tenant_key = ( message_source.source_platform_object.header.tenant_key if message_source.source_platform_object else None ) app_access_token = self.get_app_access_token() tenant_access_token = self.get_tenant_access_token(tenant_key) req_opt: RequestOption = ( RequestOption.builder() .app_ticket(self.app_ticket) .tenant_key(tenant_key) .app_access_token(app_access_token) .tenant_access_token(tenant_access_token) .build() ) response: ReplyMessageResponse = await self.api_client.im.v1.message.areply(request, req_opt) if not response.success(): raise Exception( f'client.im.v1.message.reply 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)}' ) # Send media messages separately (image, audio, file, etc.) for media in media_items: request: ReplyMessageRequest = ( ReplyMessageRequest.builder() .message_id(message_source.message_chain.message_id) .request_body( ReplyMessageRequestBody.builder() .content(json.dumps(media['content'])) .msg_type(media['msg_type']) .reply_in_thread(False) .uuid(str(uuid.uuid4())) .build() ) .build() ) tenant_key = ( message_source.source_platform_object.header.tenant_key if message_source.source_platform_object else None ) app_access_token = self.get_app_access_token() tenant_access_token = self.get_tenant_access_token(tenant_key) req_opt: RequestOption = ( RequestOption.builder() .app_ticket(self.app_ticket) .tenant_key(tenant_key) .app_access_token(app_access_token) .tenant_access_token(tenant_access_token) .build() ) response: ReplyMessageResponse = await self.api_client.im.v1.message.areply(request, req_opt) if not response.success(): raise Exception( f'client.im.v1.message.reply ({media["msg_type"]}) 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)}' ) async def reply_message_chunk( self, message_source: platform_events.MessageEvent, bot_message, message: platform_message.MessageChain, quote_origin: bool = False, is_final: bool = False, ): """ 回复消息变成更新卡片消息 Supports Dify form-action resume: when the runner yields a chunk with ``_resume_from_form=True``, the card transitions from buttons to a grey selection notice and a new ``streaming_txt_resume`` element is added for subsequent resume chunks to stream into. When ``_open_new_card=True`` on the final chunk, the existing card is left as-is and the pipeline will create a new card (with fresh form buttons) for the re-pause. """ message_id = bot_message.resp_message_id msg_seq = bot_message.msg_sequence form_data = getattr(bot_message, '_form_data', None) resume_from = getattr(bot_message, '_resume_from_form', False) action_title = getattr(bot_message, '_resume_action_title', '') resume_node_title = getattr(bot_message, '_resume_node_title', '') open_new_card = getattr(bot_message, '_open_new_card', False) # ── decide whether this chunk needs a card update ──────────────────── card_id = self.card_id_dict.get(message_id) if not card_id: return if action_title: # Build the selected notice with node_title, form_content, and action notice_parts = [] if resume_node_title: notice_parts.append(f'**{resume_node_title}**') stored_form_content = self.card_form_content.get(card_id, '') if stored_form_content: notice_parts.append(stored_form_content) completed_lines = _lark_completed_input_lines( { 'input_defs': self.card_form_input_defs.get(card_id, []), 'inputs': self.card_form_inputs.get(card_id, {}), } ) if completed_lines and not all(line in stored_form_content for line in completed_lines): notice_parts.append('---\n' + '\n'.join(completed_lines)) notice_parts.append(f'---\n✅ {action_title}') selected_notice = '\n\n'.join(notice_parts) else: selected_notice = '' # ── convert message chain → text ───────────────────────────────────── text_elements, media_items = await self.message_converter.yiri2target(message, self.api_client) text_message = '' if text_elements: parts = [] for paragraph in text_elements: para_text = ''.join(ele['text'] for ele in paragraph if ele['tag'] in ('text', 'md')) if para_text: parts.append(para_text) text_message = '\n\n'.join(parts) tenant_key = ( message_source.source_platform_object.header.tenant_key if message_source.source_platform_object else None ) app_access_token = self.get_app_access_token() tenant_access_token = self.get_tenant_access_token(tenant_key) req_opt: RequestOption = ( RequestOption.builder() .app_ticket(self.app_ticket) .tenant_key(tenant_key) .app_access_token(app_access_token) .tenant_access_token(tenant_access_token) .build() ) card_sequence = self._next_card_sequence(card_id, msg_seq) # ── RESUME: first chunk after button click ─────────────────────────── if resume_from and card_id not in self.card_resume_transitioned: # Transition the card from the form state into resume mode. # Preserve the text that was shown before the pause, and seed the # resume placeholder with the current resume content if we already # have any on the first yielded chunk. pre_pause_text = self.card_pre_pause_text.get(card_id) or self.card_streaming_text.get(card_id, '') initial_resume_text = text_message or '\u200b' await self._update_card_layout( card_id=card_id, message_source=message_source, text_message=pre_pause_text, sequence=card_sequence, form_data=None, notice_text=selected_notice, resume_placeholder_text=initial_resume_text, ) self.card_resume_transitioned.add(card_id) self.card_pre_pause_text[card_id] = pre_pause_text self.card_streaming_text[card_id] = text_message if not is_final: return # ── RESUME: subsequent chunks → full card update ───────────────────── if resume_from and card_id in self.card_resume_transitioned: cached = self.card_streaming_text.get(card_id, '') if text_message != cached: self.card_streaming_text[card_id] = text_message pre_pause_text = self.card_pre_pause_text.get(card_id, '') await self._update_card_layout( card_id=card_id, message_source=message_source, text_message=pre_pause_text, sequence=card_sequence, form_data=None, notice_text=selected_notice, resume_placeholder_text=text_message, ) if not is_final: return # ── NORMAL streaming (non-resume): update streaming_txt in-place ────── if _lark_should_update_stream_element( resume_from=resume_from, form_data=form_data, msg_seq=msg_seq, is_final=is_final, ): cached = self.card_streaming_text.get(card_id) if text_message != cached: self.card_streaming_text[card_id] = text_message request: ContentCardElementRequest = ( ContentCardElementRequest.builder() .card_id(card_id) .element_id('streaming_txt') .request_body( ContentCardElementRequestBody.builder().content(text_message).sequence(card_sequence).build() ) .build() ) response: ContentCardElementResponse = await self.api_client.cardkit.v1.card_element.acontent( request, req_opt ) if not response.success(): raise Exception( f'client.cardkit.v1.card_element.acontent failed, code: {response.code}, ' f'msg: {response.msg}, log_id: {response.get_log_id()}, ' f'resp: \n{json.dumps(json.loads(response.raw.content), indent=4, ensure_ascii=False)}' ) # ── FINAL chunk: full card layout update ───────────────────────────── if is_final: final_seq = self._next_card_sequence(card_id, card_sequence + 1) pre_pause = self.card_pre_pause_text.get(card_id, text_message) resume_cached = self.card_streaming_text.get(card_id, '') if form_data: if open_new_card: # The old card has already been laid out into resume mode # by the resume-transition block above (notice + resume # placeholder). Finalise it as a frozen step snapshot and # spawn a brand-new card to host the next human-input # prompt — each step stays visible as its own card in the # chat history. new_card_id = await self._open_new_form_card(message_id, message_source, form_data) if new_card_id is None: # Fallback: keep the existing in-place behaviour so the # workflow remains continuable even if creating the # new card failed. await self._update_card_layout( card_id=card_id, message_source=message_source, text_message=pre_pause, sequence=final_seq, form_data=form_data, resume_placeholder_text=resume_cached, show_form_prompt=True, ) self.card_streaming_text.pop(card_id, None) self.card_pre_pause_text.pop(card_id, None) self.card_form_content.pop(card_id, None) self.card_form_input_defs.pop(card_id, None) self.card_form_inputs.pop(card_id, None) else: # The old card is now a frozen snapshot; let go of its # streaming-side state but keep its source registrations # intact (no _drop_card_state) so historical button # callbacks aimed at it can still be matched if needed. self.card_streaming_text.pop(card_id, None) self.card_pre_pause_text.pop(card_id, None) self.card_form_content.pop(card_id, None) self.card_form_input_defs.pop(card_id, None) self.card_form_inputs.pop(card_id, None) self.card_resume_transitioned.discard(card_id) else: # Initial pause path: render prompt + buttons in place on # the current card. await self._update_card_layout( card_id=card_id, message_source=message_source, text_message=text_message, sequence=final_seq, form_data=form_data, show_form_prompt=True, ) # Preserve the pre-pause text so the main content can be # restored when the user clicks a button and the card # transitions to resume mode. self.card_pre_pause_text[card_id] = self.card_streaming_text.get(card_id, '') self.card_streaming_text[card_id] = '' # Store cleaned form state for the resume notice. self.card_form_content[card_id] = _lark_visible_form_content(form_data) self.card_form_input_defs[card_id] = _lark_form_input_defs(form_data) self.card_form_inputs[card_id] = dict(form_data.get('inputs') or {}) else: # Normal finish: remove buttons/notice and finalize the card. main_text, resume_text = _lark_final_layout_texts( resume_from=resume_from, text_message=text_message, pre_pause_cached=self.card_pre_pause_text.get(card_id), resume_cached=resume_cached, ) await self._update_card_layout( card_id=card_id, message_source=message_source, text_message=main_text, sequence=final_seq, form_data=None, notice_text=selected_notice if resume_from else '', resume_placeholder_text=resume_text, ) self._drop_card_state(card_id) self.card_id_dict.pop(message_id, None) # ── media (images / files) appended at the end ─────────────────────── if is_final and media_items: for media in media_items: media_request: ReplyMessageRequest = ( ReplyMessageRequest.builder() .message_id(message_source.message_chain.message_id) .request_body( ReplyMessageRequestBody.builder() .content(json.dumps(media['content'])) .msg_type(media['msg_type']) .reply_in_thread(False) .uuid(str(uuid.uuid4())) .build() ) .build() ) media_response: ReplyMessageResponse = await self.api_client.im.v1.message.areply( media_request, req_opt ) if not media_response.success(): raise Exception( f'client.im.v1.message.reply ({media["msg_type"]}) failed, code: {media_response.code}, msg: {media_response.msg}, log_id: {media_response.get_log_id()}' ) async def _add_form_buttons_to_card( self, card_id: str, message_source: platform_events.MessageEvent, form_data: dict, text_message: str = '', sequence: int = 1, ): """Update the entire card to include form action buttons. Uses card.aupdate to replace the card JSON with a template that includes the streaming text content plus interactive buttons. """ await self._update_card_layout( card_id=card_id, message_source=message_source, text_message=text_message, sequence=sequence, form_data=form_data, ) async def _remove_form_buttons_from_card( self, card_id: str, message_source: platform_events.MessageEvent, text_message: str = '', sequence: int = 1, ): """Replace the human-input card layout with the plain final layout.""" await self._update_card_layout( card_id=card_id, message_source=message_source, text_message=text_message, sequence=sequence, form_data=None, ) def _build_lark_form_field_elements(self, form_data: dict) -> tuple[list[dict], dict[str, str], list[str]]: elements: list[dict] = [] input_name_map: dict[str, str] = {} file_help_lines: list[str] = [] for idx, field in enumerate(_lark_current_input_defs(form_data), start=1): field_name = _dify_field_name(field) if not field_name: continue field_type = _dify_field_type(field) if field_type == 'select': options = _dify_select_options(field) component_name = _lark_form_component_name('Select', field_name, idx) input_name_map[component_name] = field_name elements.append( { 'tag': 'select_static', 'name': component_name, 'label': {'tag': 'plain_text', 'content': field_name}, 'placeholder': {'tag': 'plain_text', 'content': '请选择'}, 'options': [ { 'text': {'tag': 'plain_text', 'content': option}, 'value': option, } for option in options ], 'type': 'default', 'width': 'fill', 'required': False, } ) elif field_type in {'file', 'file-list'}: allowed_types = ', '.join(field.get('allowed_file_types') or []) allowed = f' ({allowed_types})' if allowed_types else '' if field_type == 'file-list': limit = field.get('number_limits') suffix = f', up to {limit}' if limit else '' file_help_lines.append( f'- {field_name}: upload file(s){allowed}{suffix} in chat or reply `{field_name}: `' ) else: file_help_lines.append( f'- {field_name}: upload a file{allowed} in chat or reply `{field_name}: `' ) else: component_name = _lark_form_component_name('Input', field_name, idx) input_name_map[component_name] = field_name is_multiline = field_type in {'paragraph', 'long_text', 'multiline_text', 'textarea'} input_element = { 'tag': 'input', 'name': component_name, 'label': {'tag': 'plain_text', 'content': field_name}, 'placeholder': {'tag': 'plain_text', 'content': '请输入'}, 'default_value': _dify_default_value(field), 'width': 'fill', 'required': False, } if is_multiline: input_element.update( { 'input_type': 'multiline_text', 'rows': 3, 'auto_resize': True, 'max_rows': 6, } ) elements.append(input_element) return elements, input_name_map, file_help_lines async def _update_card_layout( self, card_id: str, message_source: platform_events.MessageEvent, text_message: str = '', sequence: int = 1, form_data: dict | None = None, notice_text: str = '', resume_placeholder_text: str = '', show_form_prompt: bool = True, ): """Update the entire card layout. • form_data → show interactive buttons (initial Dify pause) • notice_text → replace buttons with a grey selection notice (resume transition) • resume_placeholder_text → add a streaming_txt_resume markdown element """ form_data = form_data or {} actions = form_data.get('actions', []) form_token = form_data.get('form_token', '') workflow_run_id = form_data.get('workflow_run_id', '') pipeline_uuid = form_data.get('pipeline_uuid', '') node_title = form_data.get('node_title', '') or 'Human Input Required' form_content = form_data.get('form_content', '') input_defs = _lark_form_input_defs(form_data) # When form_data is set, the visible content is rendered inside the # interactive container, so the top streaming text should stay empty # to avoid duplicate text above the action area. # # For resume notice state, keep the existing text visible in the card # and only add the grey "selected" notice below it. if form_data: render_text_message = '' else: render_text_message = text_message # Determine session key from message source if isinstance(message_source, platform_events.GroupMessage): session_key = f'group_{message_source.group.id}' else: session_key = f'person_{message_source.sender.id}' # Build button elements matching the existing card template's thumbsup/down format action_buttons = [] form_field_elements, input_name_map, file_help_lines = self._build_lark_form_field_elements(form_data) uses_form_container = bool(form_field_elements or input_name_map) if form_data: form_content = _lark_visible_form_content(form_data) self.card_form_content[card_id] = form_content self.card_form_input_defs[card_id] = input_defs self.card_form_inputs[card_id] = dict(form_data.get('inputs') or {}) is_field_step = bool(form_data.get('_current_input_field')) and not form_data.get('_action_select_only') if is_field_step: actions = ( [{'_input_progress': True, 'id': '', 'title': 'Next', 'button_style': 'primary'}] if uses_form_container else [] ) for action in actions: action_id = action.get('id', '') action_title = action.get('title', action_id) button_style = action.get('button_style', 'default') if button_style == 'primary': lark_button_type = 'primary' elif button_style == 'danger': lark_button_type = 'danger' else: lark_button_type = 'default' button = { 'tag': 'button', 'text': {'tag': 'plain_text', 'content': action_title}, 'type': lark_button_type, 'width': 'fill', 'size': 'medium', 'hover_tips': {'tag': 'plain_text', 'content': action_title}, 'behaviors': [ { 'type': 'callback', 'value': { 'form_action': True, 'form_token': form_token, 'workflow_run_id': workflow_run_id, 'pipeline_uuid': pipeline_uuid, 'action_id': action_id, 'session_key': session_key, 'card_id': card_id, 'input_name_map': input_name_map, '_input_progress': bool(action.get('_input_progress')), }, } ], 'margin': '0px 0px 0px 0px', } if uses_form_container: button['name'] = _lark_form_component_name('Button', action_id or action_title, len(action_buttons) + 1) button['form_action_type'] = 'submit' action_buttons.append(button) interactive_elements = [] if form_data: if show_form_prompt: interactive_elements = [ { 'tag': 'markdown', 'content': f'**[Human Input Required] {node_title}**', 'text_align': 'left', 'text_size': 'normal', 'margin': '0px 0px 4px 0px', } ] if form_content: interactive_elements.append( { 'tag': 'markdown', 'content': form_content, 'text_align': 'left', 'text_size': 'normal', 'margin': '0px 0px 8px 0px', } ) completed_lines = ( [] if form_data.get('_action_select_only') else _lark_completed_input_lines( { 'input_defs': input_defs, 'inputs': form_data.get('inputs') or {}, } ) ) if completed_lines: interactive_elements.append( { 'tag': 'markdown', 'content': '---\n' + '\n'.join(completed_lines), 'text_align': 'left', 'text_size': 'normal', 'margin': '0px 0px 8px 0px', } ) if file_help_lines: interactive_elements.append( { 'tag': 'markdown', 'content': '\n'.join(file_help_lines), 'text_align': 'left', 'text_size': 'normal', 'text_color': 'grey', 'margin': '0px 0px 8px 0px', } ) if action_buttons: interactive_elements.append( { 'tag': 'column_set', 'horizontal_spacing': '8px', 'horizontal_align': 'left', 'margin': '0px 0px 0px 0px', 'columns': [ { 'tag': 'column', 'width': 'weighted', 'elements': [btn], 'padding': '0px 0px 0px 0px', } for btn in action_buttons ], } ) # Build the full card JSON with buttons, same structure as create_card_id # ── mid_section: either form buttons, resume notice, or empty ── mid_section_elements = [] if form_data: if uses_form_container: form_elements = interactive_elements[:-1] if action_buttons else interactive_elements[:] form_elements.extend(form_field_elements) if action_buttons: form_elements.append(interactive_elements[-1]) mid_section_elements = [ { 'tag': 'form', 'name': _lark_form_component_name('Form', form_token or workflow_run_id or card_id, 1), 'direction': 'vertical', 'vertical_spacing': '12px', 'margin': '12px 0px 8px 0px', 'padding': '12px 12px 12px 12px', 'elements': form_elements, }, {'tag': 'hr', 'margin': '0px 0px 0px 0px'}, ] else: mid_section_elements = [ { 'tag': 'interactive_container', 'margin': '12px 0px 8px 0px', 'padding': '12px 12px 12px 12px', 'has_border': True, 'elements': interactive_elements, }, {'tag': 'hr', 'margin': '0px 0px 0px 0px'}, ] elif notice_text: mid_section_elements = [ { 'tag': 'markdown', 'content': notice_text, 'text_align': 'left', 'text_size': 'normal', 'margin': '8px 0px 4px 0px', 'text_color': 'grey', }, {'tag': 'hr', 'margin': '0px 0px 0px 0px'}, ] # ── resume placeholder element (empty, filled via acontent on each chunk) ── resume_elements = [] if resume_placeholder_text: resume_elements = [ { 'tag': 'markdown', 'content': resume_placeholder_text, 'text_align': 'left', 'text_size': 'normal', 'margin': '0px 0px 0px 0px', 'element_id': 'streaming_txt_resume', }, ] card_data = { 'schema': '2.0', 'config': { 'update_multi': True, 'streaming_mode': False, }, 'body': { 'direction': 'vertical', 'padding': '12px 12px 12px 12px', 'elements': [ { 'tag': 'div', 'text': { 'tag': 'plain_text', 'content': 'LangBot', 'text_size': 'normal', 'text_align': 'left', 'text_color': 'default', }, 'icon': { 'tag': 'custom_icon', 'img_key': 'img_v3_02p3_05c65d5d-9bad-440a-a2fb-c89571bfd5bg', }, }, { 'tag': 'markdown', 'content': render_text_message, 'text_align': 'left', 'text_size': 'normal', 'margin': '0px 0px 0px 0px', 'element_id': 'streaming_txt', }, *mid_section_elements, *resume_elements, { 'tag': 'column_set', 'horizontal_spacing': '12px', 'horizontal_align': 'right', 'columns': [ { 'tag': 'column', 'width': 'weighted', 'elements': [ { 'tag': 'markdown', 'content': '以上内容由 AI 生成，仅供参考。更多详细、准确信息可点击引用链接查看', 'text_align': 'left', 'text_size': 'notation', 'margin': '4px 0px 0px 0px', 'icon': { 'tag': 'standard_icon', 'token': 'robot_outlined', 'color': 'grey', }, } ], 'padding': '0px 0px 0px 0px', 'direction': 'vertical', 'horizontal_spacing': '8px', 'vertical_spacing': '8px', 'horizontal_align': 'left', 'vertical_align': 'top', 'margin': '0px 0px 0px 0px', 'weight': 1, }, *( [] if form_data else [ { 'tag': 'column', 'width': '20px', 'elements': [ { 'tag': 'button', 'text': {'tag': 'plain_text', 'content': ''}, 'type': 'text', 'width': 'fill', 'size': 'medium', 'icon': {'tag': 'standard_icon', 'token': 'thumbsup_outlined'}, 'hover_tips': {'tag': 'plain_text', 'content': '有帮助'}, 'behaviors': [{'type': 'callback', 'value': {'feedback': '有帮助'}}], 'margin': '0px 0px 0px 0px', } ], 'padding': '0px 0px 0px 0px', 'direction': 'vertical', 'horizontal_spacing': '8px', 'vertical_spacing': '8px', 'horizontal_align': 'left', 'vertical_align': 'top', 'margin': '0px 0px 0px 0px', }, { 'tag': 'column', 'width': '30px', 'elements': [ { 'tag': 'button', 'text': {'tag': 'plain_text', 'content': ''}, 'type': 'text', 'width': 'default', 'size': 'medium', 'icon': {'tag': 'standard_icon', 'token': 'thumbdown_outlined'}, 'hover_tips': {'tag': 'plain_text', 'content': '无帮助'}, 'behaviors': [{'type': 'callback', 'value': {'feedback': '无帮助'}}], 'margin': '0px 0px 0px 0px', } ], 'padding': '0px 0px 0px 0px', 'vertical_spacing': '8px', 'horizontal_align': 'left', 'vertical_align': 'top', 'margin': '0px 0px 0px 0px', }, ] ), ], 'margin': '0px 0px 4px 0px', }, ], }, } try: tenant_key = ( message_source.source_platform_object.header.tenant_key if message_source.source_platform_object else None ) app_access_token = self.get_app_access_token() tenant_access_token = self.get_tenant_access_token(tenant_key) req_opt: RequestOption = ( RequestOption.builder() .app_ticket(self.app_ticket) .tenant_key(tenant_key) .app_access_token(app_access_token) .tenant_access_token(tenant_access_token) .build() ) request: UpdateCardRequest = ( UpdateCardRequest.builder() .card_id(card_id) .request_body( UpdateCardRequestBody.builder() .sequence(sequence) .uuid(str(uuid.uuid4())) .card(Card.builder().type('card_json').data(json.dumps(card_data)).build()) .build() ) .build() ) response: UpdateCardResponse = await self.api_client.cardkit.v1.card.aupdate(request, req_opt) if not response.success(): await self.logger.error( f'Failed to update lark card with form buttons: code={response.code}, msg={response.msg}, ' f'log_id={response.get_log_id()}, resp={getattr(getattr(response, "raw", None), "content", None)}' ) except Exception: await self.logger.error(f'Error updating lark card with form buttons: {traceback.format_exc()}') async def is_muted(self, group_id: int) -> bool: return False def register_listener( self, event_type: typing.Type[platform_events.Event], callback: typing.Callable[ [platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None ], ): self.listeners[event_type] = callback def unregister_listener( self, event_type: typing.Type[platform_events.Event], callback: typing.Callable[ [platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None ], ): self.listeners.pop(event_type) def set_bot_uuid(self, bot_uuid: str): """设置 bot UUID（用于生成 webhook URL）""" self.bot_uuid = bot_uuid def get_event_type(self, data): schema = '1.0' if 'schema' in data: schema = data['schema'] if '2.0' == schema: return data['header']['event_type'] elif 'event' in data: return data['event']['type'] else: return data['type'] async def handle_unified_webhook(self, bot_uuid: str, path: str, request): """处理统一 webhook 请求。 Args: bot_uuid: Bot 的 UUID path: 子路径（如果有的话） request: Quart Request 对象 Returns: 响应数据 """ try: data = await request.json if 'encrypt' in data: encrypted = data['encrypt'] data = await asyncio.to_thread(lambda: json.loads(self.cipher.decrypt_string(encrypted))) type = self.get_event_type(data) context = EventContext(data) if 'url_verification' == type: # todo 验证verification token return {'challenge': data.get('challenge')} elif 'app_ticket' == type: self.app_ticket = context.event['app_ticket'] elif 'im.message.receive_v1' == type: try: p2v1 = P2ImMessageReceiveV1() p2v1.header = context.header event = P2ImMessageReceiveV1Data() event.message = EventMessage(context.event['message']) event.sender = EventSender(context.event['sender']) p2v1.event = event p2v1.schema = context.schema event = await self.event_converter.target2yiri(p2v1, self.api_client) except Exception: await self.logger.error(f'Error in lark callback: {traceback.format_exc()}') if event.__class__ in self.listeners: await self.listeners[event.__class__](event, self) elif 'card.action.trigger' == type: try: event_data = data.get('event', {}) operator = event_data.get('operator', {}) action = event_data.get('action', {}) context_data = event_data.get('context', {}) action_value_obj = _lark_mapping_from_value(action.get('value', {})) action_value = action_value_obj.get('feedback', '') if isinstance(action_value_obj, dict) else '' if isinstance(action_value_obj, dict) and action_value_obj.get('form_action'): form_token = action_value_obj.get('form_token', '') workflow_run_id = action_value_obj.get('workflow_run_id', '') action_id = action_value_obj.get('action_id', '') session_key = action_value_obj.get('session_key', '') form_inputs = _lark_extract_action_form_inputs(action, action_value_obj) if session_key.startswith('group_') or session_key.startswith('g:'): launcher_type = provider_session.LauncherTypes.GROUP launcher_id = ( session_key.split(':', 1)[1] if session_key.startswith('g:') else session_key[len('group_') :] ) else: launcher_type = provider_session.LauncherTypes.PERSON launcher_id = ( session_key.split(':', 1)[1] if session_key.startswith('p:') else session_key[len('person_') :] ) form_action_data = { 'form_token': form_token, 'workflow_run_id': workflow_run_id, 'action_id': action_id, 'user': f'{launcher_type.value}_{launcher_id}', 'inputs': form_inputs, } if action_value_obj.get('_input_progress'): form_action_data['_input_progress'] = True open_message_id = context_data.get('open_message_id') card_id = self.reply_message_card_ids.get(str(open_message_id)) if open_message_id else None if not card_id: card_id = str(action_value_obj.get('card_id') or '') if card_id and form_inputs: cached_inputs = dict(self.card_form_inputs.get(card_id) or {}) cached_inputs.update(form_inputs) self.card_form_inputs[card_id] = cached_inputs if self.ap is not None: self.ap.logger.info( f'Lark form action inputs cached: card_id={card_id} ' f'open_message_id={open_message_id} keys={list(form_inputs.keys())}' ) source_time = datetime.datetime.now() message_chain = platform_message.MessageChain( [platform_message.Plain(text=f'[Form Action: {action_id or "confirm"}]')] ) if open_message_id: message_chain.insert( 0, platform_message.Source( id=open_message_id, time=source_time, ), ) user_id = operator.get('open_id') or operator.get('user_id') or str(launcher_id) event_time = source_time.timestamp() if launcher_type == provider_session.LauncherTypes.GROUP: synthetic_event = platform_events.GroupMessage( sender=platform_entities.GroupMember( id=user_id, member_name='', permission=platform_entities.Permission.Member, group=platform_entities.Group( id=launcher_id, name='', permission=platform_entities.Permission.Member, ), ), message_chain=message_chain, time=event_time, source_platform_object=data, ) else: synthetic_event = platform_events.FriendMessage( sender=platform_entities.Friend( id=user_id, nickname='', remark='', ), message_chain=message_chain, time=event_time, source_platform_object=data, ) bot_uuid = '' pipeline_uuid = action_value_obj.get('pipeline_uuid') or None for bot in self.ap.platform_mgr.bots: if bot.adapter is self: bot_uuid = bot.bot_entity.uuid pipeline_uuid = pipeline_uuid or bot.bot_entity.use_pipeline_uuid break await self.ap.query_pool.add_query( bot_uuid=bot_uuid, launcher_type=launcher_type, launcher_id=launcher_id, sender_id=user_id, message_event=synthetic_event, message_chain=message_chain, adapter=self, pipeline_uuid=pipeline_uuid, variables={ '_dify_form_action': form_action_data, '_routed_by_rule': True, }, ) return {'toast': {'type': 'success', 'content': '操作成功'}} if action_value == '有帮助': feedback_type = 1 elif action_value == '无帮助': feedback_type = 2 else: return {'toast': {'type': 'success', 'content': '操作成功'}} user_id = operator.get('open_id') or operator.get('user_id') open_chat_id = context_data.get('open_chat_id') open_message_id = context_data.get('open_message_id') if open_chat_id: session_id = f'group_{open_chat_id}' elif user_id: session_id = f'person_{user_id}' else: session_id = None # Resolve monitoring message ID from reply message mapping monitoring_msg_id = None if open_message_id and open_message_id in self.reply_to_monitoring_msg: monitoring_msg_id = self.reply_to_monitoring_msg[open_message_id][0] feedback_event = platform_events.FeedbackEvent( feedback_id=data.get('header', {}).get('event_id', str(uuid.uuid4())), feedback_type=feedback_type, feedback_content=action_value, user_id=user_id, session_id=session_id, message_id=open_message_id, stream_id=monitoring_msg_id, source_platform_object=data, ) if platform_events.FeedbackEvent in self.listeners: await self.listeners[platform_events.FeedbackEvent](feedback_event, self) return {'toast': {'type': 'success', 'content': '感谢您的反馈'}} except Exception: await self.logger.error(f'Error in lark card action callback: {traceback.format_exc()}') return {'toast': {'type': 'error', 'content': '反馈处理失败'}} elif 'im.chat.member.bot.added_v1' == type: try: bot_added_welcome_msg = self.config.get('bot_added_welcome', '') if bot_added_welcome_msg: final_content = { 'zh_Hans': { 'title': '', 'content': [[{'tag': 'md', 'text': bot_added_welcome_msg}]], }, } chat_id = context.event['chat_id'] request: CreateMessageRequest = ( CreateMessageRequest.builder() .receive_id_type('chat_id') .request_body( CreateMessageRequestBody.builder() .receive_id(chat_id) .content(json.dumps(final_content)) .msg_type('post') .uuid(str(uuid.uuid4())) .build() ) .build() ) tenant_key = context.header.tenant_key if context.header else None app_access_token = self.get_app_access_token() tenant_access_token = self.get_tenant_access_token(tenant_key) req_opt: RequestOption = ( RequestOption.builder() .app_ticket(self.app_ticket) .tenant_key(tenant_key) .app_access_token(app_access_token) .tenant_access_token(tenant_access_token) .build() ) response: CreateMessageResponse = self.api_client.im.v1.message.create(request, req_opt) if not response.success(): raise Exception( 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)}' ) except Exception as e: print(f'im.chat.member.bot.added_v1: {e}') await self.logger.error(f'Error in lark callback: {traceback.format_exc()}') return {'code': 200, 'message': 'ok'} except Exception as e: print(f'Error in lark callback: {e}') await self.logger.error(f'Error in lark callback: {traceback.format_exc()}') return {'code': 500, 'message': 'error'} async def run_async(self): enable_webhook = self.config['enable-webhook'] if not enable_webhook: try: await self.bot._connect() except lark_oapi.ws.exception.ClientException as e: raise e except Exception as e: await self.bot._disconnect() if self.bot._auto_reconnect: await self.bot._reconnect() else: raise e else: # 统一 webhook 模式下，不启动独立的 Quart 应用 # 保持运行但不启动独立端口 async def keep_alive(): while True: await asyncio.sleep(1) await keep_alive() async def kill(self) -> bool: # 需要断开连接，不然旧的连接会继续运行，导致飞书消息来时会随机选择一个连接 # 断开时lark.ws.Client的_receive_message_loop会打印error日志: receive message loop exit。然后进行重连， # 所以要设置_auto_reconnect=False,让其不重连。 self.bot._auto_reconnect = False await self.bot._disconnect() inbound_tasks = list(self.inbound_event_tasks) for task in inbound_tasks: if not task.done(): task.cancel() if inbound_tasks: await asyncio.gather(*inbound_tasks, return_exceptions=True) self.inbound_event_tasks.clear() with self.threadsafe_event_lock: threadsafe_futures = list(self.threadsafe_event_futures) for future in threadsafe_futures: future.cancel() if threadsafe_futures: await asyncio.gather( *(asyncio.wrap_future(future) for future in threadsafe_futures), return_exceptions=True, ) with self.threadsafe_event_lock: self.threadsafe_event_futures.clear() self.tenant_access_tokens.clear() self.card_id_dict.clear() self.pending_monitoring_msg.clear() self.reply_to_monitoring_msg.clear() for card_id in tuple(self.card_last_accessed): self._drop_card_state(card_id) self.card_last_accessed.clear() self.reply_message_card_ids.clear() self.card_sequence_dict.clear() self.card_id_to_source_ids.clear() self.card_streaming_text.clear() self.card_pre_pause_text.clear() self.card_form_content.clear() self.card_form_input_defs.clear() self.card_form_inputs.clear() self.card_resume_transitioned.clear() return False