From 97b18e9f6694208c9554e4ec446e7f7cab6cb349 Mon Sep 17 00:00:00 2001 From: tsukasa Date: Thu, 2 Nov 2023 10:40:37 +0800 Subject: [PATCH] email parser document --- doc/tob email.md | 62 +++++++ .../knowledge_pipelines/Bucky.Mail/input.py | 156 ++++++++++++++++++ .../knowledge_manager/input/email.py | 7 +- src/component/knowledge_manager/pipeline.py | 4 +- src/knowledge/pipeline.py | 5 + 5 files changed, 226 insertions(+), 8 deletions(-) create mode 100644 doc/tob email.md create mode 100644 rootfs/knowledge_pipelines/Bucky.Mail/input.py diff --git a/doc/tob email.md b/doc/tob email.md new file mode 100644 index 0000000..f59124e --- /dev/null +++ b/doc/tob email.md @@ -0,0 +1,62 @@ +# issue tree +最核心的机制是树状的issue管理，一个issue应当包含以下属性： ++ 谁提出来的 ++ 分配给谁的，如果有的话 ++ 起始日期 ++ deadline，如果有的话 ++ 在哪个邮件里面提出的，引用某个email的原始链接 ++ 这个issue的summary，有几种情况， + + 一个新的任务，要达成什么目标 + + 提出了一个问题，需求答案 + + 解决了某个issue，完成了task或者解答了一个问题 ++ 推断出来的 issue的状态，进行中，关闭，超时，完成了 ++ parent issue + +knowledge维护一个issue tree，从一个root issue出发（root可以是抽象的，比如一个组织的存在，并不是具体的）；knowledge env 提供对这个issue tree的维护接口： ++ 新增issue ++ 更新issue + +# parse email +假定从从某个起始日期开始，以每天为单位，扫描当天新增的email，对每封email： +1. 输入email 和 从knowledge base获取 issue tree +2. llm提示词应当包括：issue tree， email正文， knowledge env， llm完成如下推理： ++ email正文提出了一个新的issue，在knowledge env新增issue ++ email正文改变了一个issue的状态 + + 通报完成了一个task + + 回答了一个问题 + + 明确改变一个issue的状态：认为完成，要延期，认为要取消 ++ 根据推理结果正确产生knowledge env 的调用，更新issue tree的状态 + +## 推理部分可能的out of token： +1. 裁剪掉已经关闭，超时的 issue +2. 根据标题特征，是不是对某个email的回复，定位到某个issue， 裁剪出 sub tree +2. 很长的邮件正文： + 1. 第一种方法：先llm推理email的summary，再把summary当正文输入推理issue + 2. 第二种方法（我觉得更好）：分片迭代输入email正文，单次llm推理的提示词就变成：issue tree， 当前email summary， 当段email正文，knowledge env： + + env里面新增一个method，更新当前email summary + + +# build issue tree +## 第一种结构：基于knowledge pipeline +1. pipeline input： 判定当前时间晚于 起始时间并且早于下一个自然天，开始爬正确范围内的邮件输入 +2. pipeline parser：包含准备user prompt 的计算部分，和几个agent ++ 计算部分： 裁剪issue tree，[可选的：调用llm推理生成summary] ++ agent 部分： + + agent提示词：从输入的结构化issue tree， 和邮件正文，回复对issue tree knowledge env的调用 + + 输入提示词： email 正文或者summary，裁剪后的issue tree ++ parser的流程： + 对每一个输入的email，查询（裁剪）当前issue tree，把email 和 issue tree 当作user prompt发送给agent，等待agent返回 + + +## 第二种结构：基于agent workspace（待定） +1. schedule task：在每一天产生一个build issue tree task +2. build issue tree agent: 响应build issue tree task（可不可以以计算为入口，还是只能agent入口） ++ agent调用email env，读出一封邮件 ++ agent调用knowledge env，返回issue tree ++ agent从邮件内容和issue tree推理，回复对issue tree knowledge env 的调用 + +# query issue tree +主动的或者被动的根据当前issue tree的状态，推理出一些汇总的结论： ++ 是不是有超期的事项 ++ 事情是不是有在推进 ++ 有哪些事情完成了 \ No newline at end of file diff --git a/rootfs/knowledge_pipelines/Bucky.Mail/input.py b/rootfs/knowledge_pipelines/Bucky.Mail/input.py new file mode 100644 index 0000000..23ae40a --- /dev/null +++ b/rootfs/knowledge_pipelines/Bucky.Mail/input.py @@ -0,0 +1,156 @@ +import os +import aiofiles +import chardet +import logging +import string +from knowledge import ImageObjectBuilder, DocumentObjectBuilder, KnowledgePipelineEnvironment, KnowledgePipelineJournal +from aios_kernel.storage import AIStorage + +class KnowledgeEmailSource: + def __init__(self, env: KnowledgePipelineEnvironment, config:dict): + self.config = config + + # @classmethod + # def user_config_items(cls): + # return [("address", "email address"), + # ("password", "email password"), + # ("imap_server", "imap server"), + # ("imap_port", "imap port") + # ] + + async def run_once(self): + # read config from toml file + # and read from config config.local.toml if exists (config.local.toml is ignored by git) + logging.debug(f"knowledge email source {self.id()} run once") + filter = "ALL" + self.client = self.email_client() + await self.read_emails(imap_keyword=filter) + + def email_client(self) -> imaplib.IMAP4_SSL: + logging.info(f"read email config from {self.config.get('imap_server')}") + client = imaplib.IMAP4_SSL( + host=self.config.get('imap_server'), + port=self.config.get('imap_port') + ) + client.login(self.config.get('address'), self.config.get('password')) + return client + + async def read_emails(self, folder: str = 'INBOX', imap_keyword: str = "UNSEEN"): + journal_client = KnowledgeJournalClient() + latest_journal = journal_client.latest_journal(self.id()) + latest_uid = 0 if latest_journal is None else int(latest_journal.item_id) + self.client.select(folder) + _, data = self.client.uid('search', None, imap_keyword) + + # get email uid list + email_list = data[0].split() + logging.info(f"got {len(email_list)} emails") + journal_client = KnowledgeJournalClient() + for uid in email_list: + _uid = int.from_bytes(uid) + if _uid > latest_uid: + email_dir = self.check_email_saved(uid) + if email_dir is not None: + logging.info(f"email uid {uid} already saved") + else: + email_dir = self.read_and_save_email(uid) + logging.info(f"email uid {uid} saved") + email_object = EmailObjectBuilder({}, email_dir).build() + await KnowledgeBase().insert_object(email_object) + journal_client.insert(KnowledgeJournal("email", self.id(), str(int.from_bytes(uid)), str(email_object.calculate_id()))) + + + def read_and_save_email(self, uid: str) -> str: + message_parts = "(BODY.PEEK[])" + _, email_data = self.client.uid('fetch', uid, message_parts) + mail = mailparser.parse_from_bytes(email_data[0][1]) + logging.info(f"got email subject [{mail.subject}]") + self.save_email(mail) + return self.get_local_dir_name(mail) + + def get_local_dir_name(self, mail: mailparser.MailParser) -> str: + dir = f"{self.local_root()}/{self.config.get('address')}" + name = f"{mail.subject}__{mail.date}" + name = hashlib.md5(name.encode('utf-8')).hexdigest() + return f"{dir}/{name}" + + def check_email_saved(self, uid: str) -> str: + message_parts = "(BODY[HEADER])" + _, email_data = self.client.uid('fetch', uid, message_parts) + mail = mailparser.parse_from_bytes(email_data[0][1]) + logging.info(f"[{uid}]check email subject [{mail.subject}]") + dir = self.get_local_dir_name(mail) + logging.info(f"check email saved {dir}") + file = f"{dir}/email.txt" + if os.path.exists(file): + return dir + return None + + # save email attachment(images) + def save_email_attachment(self, mail: mailparser.MailParser, email_dir: str): + for attachment in mail.attachments: + if attachment['mail_content_type'] in ['image/png', 'image/jpeg', 'image/gif']: + print('current mail have image attachment') + img_dir = f"{email_dir}/image" + if not os.path.exists(img_dir): + os.makedirs(img_dir) + filename = attachment['filename'] + filefullname = f"{img_dir}/{filename}" + image_data = attachment['payload'] + try: + image_data = base64.b64decode(image_data) + except base64.binascii.Error: + image_data = image_data.encode() + with open(filefullname, 'wb') as f: + f.write(image_data) + logging.info(f"save email image {filename} success") + + # save email body images(html content) + def save_body_images(self, html_content: str, email_dir: str): + # get all image urls + soup = BeautifulSoup(html_content, 'html.parser') + img_tags = soup.find_all('img') + img_urls = [img['src'] for img in img_tags if 'src' in img.attrs] + logging.info(f'Found {len(img_urls)} images in email body') + + name_count = 0 + + if not os.path.exists(email_dir): + os.makedirs(email_dir) + + for img_url in img_urls: + # keep the original image filename(last of url) + ext = img_url.split('/')[-1].split('.')[-1] + img_filename = os.path.join(email_dir, f"{name_count}.{ext}") + name_count += 1 + # download image + response = requests.get(img_url, stream=True) + if response.status_code == 200: + with open(img_filename, 'wb') as img_file: + for chunk in response.iter_content(1024): + img_file.write(chunk) + logging.info(f'Downloaded {img_url} to {img_filename}') + else: + logging.info(f'Failed to download {img_url}') + + # save email content to local dir + def save_email(self, mail: mailparser.MailParser): + dir = f"{self.local_root()}/{self.config.get('address')}" + if not os.path.exists(dir): + os.makedirs(dir) + email_dir = self.get_local_dir_name(mail) + logging.info(f"save email to {email_dir}") + if not os.path.exists(email_dir): + os.makedirs(email_dir) + with open(f"{email_dir}/email.txt", "w", encoding='utf-8') as f: + # soup = BeautifulSoup(mail.body, 'html.parser') + f.write(mail.body) + with open(f"{email_dir}/meta.json", "w", encoding='utf-8') as f: + mail_dict = json.loads(mail.mail_json) + if 'body' in mail_dict: + del mail_dict['body'] + json.dump(mail_dict, f, ensure_ascii=False, indent=4) + logging.info(f"save email meta info {f.name}") + + self.save_email_attachment(mail, email_dir) + self.save_body_images(mail.body, f"{email_dir}/body_image") \ No newline at end of file diff --git a/src/component/knowledge_manager/input/email.py b/src/component/knowledge_manager/input/email.py index a8eb353..3bcc6b3 100644 --- a/src/component/knowledge_manager/input/email.py +++ b/src/component/knowledge_manager/input/email.py @@ -1,6 +1,6 @@ class KnowledgeEmailSource: - def __init__(self, config:dict): + def __init__(self, env: KnowledgePipelineEnvironment, config:dict): self.config = config self.config["type"] = "email" @@ -15,11 +15,6 @@ class KnowledgeEmailSource: ("imap_port", "imap port") ] - @classmethod - def local_root(cls): - user_data_dir = AIStorage.get_instance().get_myai_dir() - return os.path.abspath(f"{user_data_dir}/knowledge/email") - async def run_once(self): # read config from toml file # and read from config config.local.toml if exists (config.local.toml is ignored by git) diff --git a/src/component/knowledge_manager/pipeline.py b/src/component/knowledge_manager/pipeline.py index 55a5105..1701f4f 100644 --- a/src/component/knowledge_manager/pipeline.py +++ b/src/component/knowledge_manager/pipeline.py @@ -46,7 +46,7 @@ class KnowledgePipelineManager: input_init = runpy.run_path(input_module)["init"] else: input_init = self.input_modules.get(input_module) - input_params = config["input"]["params"] + input_params = config["input"].get("params") parser_module = config["parser"]["module"] _, ext = os.path.splitext(parser_module) @@ -55,7 +55,7 @@ class KnowledgePipelineManager: parser_init = runpy.run_path(parser_module)["init"] else: parser_init = self.parser_modules.get(parser_module) - parser_params = config["parser"]["params"] + parser_params = config["parser"].get("params") data_path = os.path.join(self.root_dir, name) diff --git a/src/knowledge/pipeline.py b/src/knowledge/pipeline.py index af3b665..62f246a 100644 --- a/src/knowledge/pipeline.py +++ b/src/knowledge/pipeline.py @@ -1,6 +1,7 @@ import datetime import sqlite3 import os +import logging from . import ObjectID, KnowledgeStore from enum import Enum @@ -66,12 +67,16 @@ class KnowledgePipelineEnvironment: os.makedirs(pipeline_path) self.pipeline_path = pipeline_path self.journal = KnowledgePipelineJournalClient(pipeline_path) + self.logger = logging.getLogger() def get_journal(self) -> KnowledgePipelineJournalClient: return self.journal def get_knowledge_store(self) -> KnowledgeStore: return self.knowledge_store + + def get_logger(self) -> logging.Logger: + return self.logger class KnowledgePipelineState(Enum): INIT = 0