diff --git a/rootfs/agents/FindPhoto/agent.toml b/rootfs/agents/FindPhoto/agent.toml deleted file mode 100644 index ddb3f3f..0000000 --- a/rootfs/agents/FindPhoto/agent.toml +++ /dev/null @@ -1,21 +0,0 @@ -instance_id = "FindPhoto" -fullname = "FindPhoto" -llm_model_name = "gpt-4" -max_token_size = 16000 -enable_timestamp = "false" -owner_prompt = "我是你的主人{name}" -contact_prompt = "我是你的朋友{name}" -owner_env = "environment.py" - -[[prompt]] -role = "system" -content = """ -你是FindPhoto,你可以访问我的照片目录。 - -*** -你在收到我的信息后，按如下规则处理 -1. 在第一次接受到一条信息时，优先尝试用合适的关键字查询去查询知识库。 -2. 如果信息中包含一段知识库的查询结果，尝试用查询结果处理，如果还是不能处理，尝试递增index继续查询。 -3. 如果要返回知识库结果条目，在消息开头附上他的json字符串。 -""" - diff --git a/rootfs/agents/FindPhoto/enviroment.py b/rootfs/agents/FindPhoto/enviroment.py deleted file mode 100644 index e69de29..0000000 diff --git a/rootfs/knowledge_pipelines/Bucky.Mail/input.py b/rootfs/knowledge_pipelines/Bucky.Mail/input.py deleted file mode 100644 index 23ae40a..0000000 --- a/rootfs/knowledge_pipelines/Bucky.Mail/input.py +++ /dev/null @@ -1,156 +0,0 @@ -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/rootfs/knowledge_pipelines/Mail/Issue/local.py b/rootfs/knowledge_pipelines/Mail/Issue/local.py new file mode 100644 index 0000000..1edfff8 --- /dev/null +++ b/rootfs/knowledge_pipelines/Mail/Issue/local.py @@ -0,0 +1,9 @@ +import sys +import os +from knowledge import KnowledgePipelineEnvironment +directory = os.path.dirname(__file__) +sys.path.append(directory + '/../../../../src/component/') + +from mail_environment import LocalEmail +def init(env: KnowledgePipelineEnvironment, params: dict): + return LocalEmail(env, params) \ No newline at end of file diff --git a/rootfs/knowledge_pipelines/Mail/Issue/parser.py b/rootfs/knowledge_pipelines/Mail/Issue/parser.py new file mode 100644 index 0000000..a5f819b --- /dev/null +++ b/rootfs/knowledge_pipelines/Mail/Issue/parser.py @@ -0,0 +1,10 @@ +import sys +import os +from knowledge import * +directory = os.path.dirname(__file__) + +sys.path.append(directory + '/../../../../src/component/') +from mail_environment import IssueParser + +def init(env: KnowledgePipelineEnvironment, params: dict): + return IssueParser(env, params) \ No newline at end of file diff --git a/rootfs/knowledge_pipelines/Mail/Issue/pipeline.toml b/rootfs/knowledge_pipelines/Mail/Issue/pipeline.toml new file mode 100644 index 0000000..d909ba9 --- /dev/null +++ b/rootfs/knowledge_pipelines/Mail/Issue/pipeline.toml @@ -0,0 +1,9 @@ +name = "Mail.Issue" +input.module = "local.py" +input.params.path = "${myai_dir}/mail" +input.params.watch = true +parser.module = "parser.py" +parser.params.mail_path = "${myai_dir}/mail" +parser.params.issue_path = "${myai_dir}/mail/issue.json" +parser.params.root_issue = "巴克云公司推进中的项目" + diff --git a/rootfs/knowledge_pipelines/Mail/Spider/input.py b/rootfs/knowledge_pipelines/Mail/Spider/input.py new file mode 100644 index 0000000..d5e6958 --- /dev/null +++ b/rootfs/knowledge_pipelines/Mail/Spider/input.py @@ -0,0 +1,10 @@ +import sys +import os +from knowledge import KnowledgePipelineEnvironment +directory = os.path.dirname(__file__) +sys.path.append(directory + '/../../../../src/component/') + +from mail_environment import EmailSpider + +def init(env: KnowledgePipelineEnvironment, params: dict): + return EmailSpider(env, params) \ No newline at end of file diff --git a/rootfs/knowledge_pipelines/Mail/Spider/pipeline.toml b/rootfs/knowledge_pipelines/Mail/Spider/pipeline.toml new file mode 100644 index 0000000..df9f306 --- /dev/null +++ b/rootfs/knowledge_pipelines/Mail/Spider/pipeline.toml @@ -0,0 +1,4 @@ +name = "Mail.Issue" +input.module = "input.py" +input.params.path = "${myai_dir}/data" + diff --git a/rootfs/knowledge_pipelines/pipelines.toml b/rootfs/knowledge_pipelines/pipelines.toml index 3c78084..7e69436 100644 --- a/rootfs/knowledge_pipelines/pipelines.toml +++ b/rootfs/knowledge_pipelines/pipelines.toml @@ -1,3 +1,3 @@ pipelines = [ - "Mia" + "Mail/Issue" ] \ No newline at end of file diff --git a/src/aios_kernel/__init__.py b/src/aios_kernel/__init__.py index 9093b07..a1ab85f 100644 --- a/src/aios_kernel/__init__.py +++ b/src/aios_kernel/__init__.py @@ -1,7 +1,7 @@ from .environment import Environment,EnvironmentEvent from .agent_base import AgentMsg,AgentMsgStatus,AgentMsgType,AgentPrompt from .chatsession import AIChatSession -from .agent import AIAgent,AIAgentTemplete +from .agent import AIAgent,AIAgentTemplete, BaseAIAgent from .compute_kernel import ComputeKernel,ComputeTask,ComputeTaskResult,ComputeTaskState,ComputeTaskType from .compute_node import ComputeNode,LocalComputeNode from .open_ai_node import OpenAI_ComputeNode diff --git a/src/aios_kernel/agent_base.py b/src/aios_kernel/agent_base.py index 95b5e9e..60c5fe6 100644 --- a/src/aios_kernel/agent_base.py +++ b/src/aios_kernel/agent_base.py @@ -11,8 +11,10 @@ import shlex import json from typing import List -from .ai_function import FunctionItem -from .compute_task import ComputeTaskResult +from .ai_function import FunctionItem, AIFunction +from .compute_task import ComputeTaskResult,ComputeTaskResultCode +from .environment import Environment + logger = logging.getLogger(__name__) @@ -592,3 +594,114 @@ class CustomAIAgent(BaseAIAgent): def get_llm_learn_token_limit(self) -> int: return self.llm_learn_token_limit + +class BaseAIAgent: + def __init__(self) -> None: + pass + + @classmethod + def _get_inner_functions(cls, env:Environment) -> (dict,int): + if env is None: + return None,0 + + all_inner_function = env.get_all_ai_functions() + if all_inner_function is None: + return None,0 + + result_func = [] + result_len = 0 + for inner_func in all_inner_function: + func_name = inner_func.get_name() + this_func = {} + this_func["name"] = func_name + this_func["description"] = inner_func.get_description() + this_func["parameters"] = inner_func.get_parameters() + result_len += len(json.dumps(this_func)) / 4 + result_func.append(this_func) + + return result_func,result_len + + @classmethod + async def do_llm_complection( + cls, + env:Environment, + prompt:AgentPrompt, + org_msg:AgentMsg, + llm_model_name:str, + max_token_size:int + ) -> ComputeTaskResult: + from .compute_kernel import ComputeKernel + #logger.debug(f"Agent {self.agent_id} do llm token static system:{system_prompt_len},function:{function_token_len},history:{history_token_len},input:{input_len}, totoal prompt:{system_prompt_len + function_token_len + history_token_len} ") + inner_functions,inner_functions_len = cls._get_inner_functions(env) + task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,llm_model_name,max_token_size,inner_functions) + if task_result.result_code != ComputeTaskResultCode.OK: + logger.error(f"llm compute error:{task_result.error_str}") + #error_resp = msg.create_error_resp(task_result.error_str) + return task_result + + result_message = task_result.result.get("message") + inner_func_call_node = None + if result_message: + inner_func_call_node = result_message.get("function_call") + + if inner_func_call_node: + call_prompt : AgentPrompt = copy.deepcopy(prompt) + task_result = await cls._execute_func(env,inner_func_call_node,call_prompt,inner_functions,org_msg,llm_model_name,max_token_size) + + return task_result + + @classmethod + async def _execute_func( + cls, + env: Environment, + inner_func_call_node: dict, + prompt: AgentPrompt, + inner_functions: dict, + org_msg:AgentMsg, + llm_model_name:str, + max_token_size:int, + stack_limit = 5 + ) -> ComputeTaskResult: + from .compute_kernel import ComputeKernel + func_name = inner_func_call_node.get("name") + arguments = json.loads(inner_func_call_node.get("arguments")) + logger.info(f"llm execute inner func:{func_name} ({json.dumps(arguments)})") + + func_node : AIFunction = env.get_ai_function(func_name) + if func_node is None: + result_str = f"execute {func_name} error,function not found" + else: + if org_msg: + ineternal_call_record = AgentMsg.create_internal_call_msg(func_name,arguments,org_msg.get_msg_id(),org_msg.target) + + try: + result_str:str = await func_node.execute(**arguments) + except Exception as e: + result_str = f"execute {func_name} error:{str(e)}" + logger.error(f"llm execute inner func:{func_name} error:{e}") + + + logger.info("llm execute inner func result:" + result_str) + + prompt.messages.append({"role":"function","content":result_str,"name":func_name}) + task_result:ComputeTaskResult = await ComputeKernel.get_instance().do_llm_completion(prompt,llm_model_name,max_token_size,inner_functions) + if task_result.result_code != ComputeTaskResultCode.OK: + logger.error(f"llm compute error:{task_result.error_str}") + return task_result + + ineternal_call_record.result_str = task_result.result_str + ineternal_call_record.done_time = time.time() + if org_msg: + org_msg.inner_call_chain.append(ineternal_call_record) + + inner_func_call_node = None + if stack_limit > 0: + result_message : dict = task_result.result.get("message") + if result_message: + inner_func_call_node = result_message.get("function_call") + + if inner_func_call_node: + return await cls._execute_func(env,inner_func_call_node,prompt,inner_functions,org_msg,llm_model_name,max_token_size,stack_limit-1) + else: + return task_result +>>>>>>> 2f9cee9 (a issue parser of email) diff --git a/src/aios_kernel/environment.py b/src/aios_kernel/environment.py index cf5e9b3..ab384cd 100644 --- a/src/aios_kernel/environment.py +++ b/src/aios_kernel/environment.py @@ -45,8 +45,8 @@ class Environment: #@abstractmethod #TODO: how to use env? different env has different prompt - #def get_env_prompt(self) -> str: - # pass + def get_env_prompt(self) -> str: + pass def add_ai_function(self,func:AIFunction) -> None: if self.functions.get(func.get_name()) is not None: diff --git a/src/component/knowledge_manager/input/email.py b/src/component/knowledge_manager/input/email.py deleted file mode 100644 index 3bcc6b3..0000000 --- a/src/component/knowledge_manager/input/email.py +++ /dev/null @@ -1,153 +0,0 @@ - -class KnowledgeEmailSource: - def __init__(self, env: KnowledgePipelineEnvironment, config:dict): - self.config = config - self.config["type"] = "email" - - def id(self): - return self.config["address"] - - @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/local_dir.py b/src/component/knowledge_manager/input/local_dir.py deleted file mode 100644 index e9a54f4..0000000 --- a/src/component/knowledge_manager/input/local_dir.py +++ /dev/null @@ -1,68 +0,0 @@ -import os -import aiofiles -import chardet -import logging -import string -from knowledge import ImageObjectBuilder, DocumentObjectBuilder, KnowledgePipelineEnvironment, KnowledgePipelineJournal -from aios_kernel.storage import AIStorage - -class KnowledgeDirSource: - def __init__(self, env: KnowledgePipelineEnvironment, config): - self.env = env - path = string.Template(config["path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir()) - config["path"] = path - self.config = config - - # @classmethod - # def user_config_items(cls): - # return [("path", "local dir path")] - - def path(self): - return self.config["path"] - - @staticmethod - async def read_txt_file(file_path:str)->str: - cur_encode = "utf-8" - async with aiofiles.open(file_path,'rb') as f: - cur_encode = chardet.detect(await f.read())['encoding'] - - async with aiofiles.open(file_path,'r',encoding=cur_encode) as f: - return await f.read() - - async def next(self): - while True: - journals = self.env.journal.latest_journals(1) - from_time = 0 - if len(journals) == 1: - latest_journal = journals[0] - if latest_journal.is_finish(): - yield None - continue - from_time = os.path.getctime(latest_journal.get_input()) - if os.path.getmtime(self.path()) <= from_time: - yield (None, None) - continue - - file_pathes = sorted(os.listdir(self.path()), key=lambda x: os.path.getctime(os.path.join(self.path(), x))) - for rel_path in file_pathes: - file_path = os.path.join(self.path(), rel_path) - timestamp = os.path.getctime(file_path) - if timestamp <= from_time: - continue - ext = os.path.splitext(file_path)[1].lower() - if ext in ['.jpg', '.jpeg', '.png', '.gif', '.bmp']: - logging.info(f"knowledge dir source found image file {file_path}") - image = ImageObjectBuilder({}, {}, file_path).build(self.env.get_knowledge_store()) - await self.env.get_knowledge_store().insert_object(image) - yield (image.calculate_id(), file_path) - if ext in ['.txt']: - logging.info(f"knowledge dir source found text file {file_path}") - text = await self.read_txt_file(file_path) - document = DocumentObjectBuilder({}, {}, text).build(self.env.get_knowledge_store()) - await self.env.get_knowledge_store().insert_object(document) - yield (document.calculate_id(), file_path) - yield (None, None) - - -def init(env: KnowledgePipelineEnvironment, params: dict) -> KnowledgeDirSource: - return KnowledgeDirSource(env, params) \ No newline at end of file diff --git a/src/component/knowledge_manager/parser/embedding.py b/src/component/knowledge_manager/parser/embedding.py deleted file mode 100644 index cb8b5f8..0000000 --- a/src/component/knowledge_manager/parser/embedding.py +++ /dev/null @@ -1,102 +0,0 @@ -# define a knowledge base class -import json -import string -from aios_kernel import ComputeKernel, AIStorage -from knowledge import * - - -class EmbeddingParser: - def __init__(self, env: KnowledgePipelineEnvironment, config: dict): - self._default_text_model = "all-MiniLM-L6-v2" - self._default_image_model = "clip-ViT-B-32" - - path = string.Template(config["path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir()) - if not os.path.exists(path): - os.makedirs(path) - config["path"] = path - - self.env = env - self.config = config - - def get_path(self) -> str: - return self.config["path"] - - def __get_vector_store(self, model_name: str) -> ChromaVectorStore: - return ChromaVectorStore(self.get_path(), model_name) - - async def __embedding_document(self, document: DocumentObject): - for chunk_id in document.get_chunk_list(): - chunk = self.env.get_knowledge_store().get_chunk_reader().get_chunk(chunk_id) - if chunk is None: - raise ValueError(f"text chunk not found: {chunk_id}") - - text = chunk.read().decode("utf-8") - vector = await ComputeKernel.get_instance().do_text_embedding(text, self._default_text_model) - if vector: - await self.__get_vector_store(self._default_text_model).insert(vector, chunk_id) - - async def __embedding_image(self, image: ImageObject): - # desc = {} - # if not not image.get_meta(): - # desc["meta"] = image.get_meta() - # if not not image.get_exif(): - # desc["exif"] = image.get_exif() - # if not not image.get_tags(): - # desc["tags"] = image.get_tags() - # vector = await self.compute_kernel.do_text_embedding(json.dumps(desc), self._default_text_model) - vector = await ComputeKernel.get_instance().do_image_embedding(image.calculate_id(), self._default_image_model) - if vector: - await self.__get_vector_store(self._default_image_model).insert(vector, image.calculate_id()) - - async def __embedding_video(self, vedio: VideoObject): - desc = {} - if not not vedio.get_meta(): - desc["meta"] = vedio.get_meta() - if not not vedio.get_info(): - desc["info"] = vedio.get_info() - if not not vedio.get_tags(): - desc["tags"] = vedio.get_tags() - vector = await ComputeKernel.get_instance().do_text_embedding(json.dumps(desc), self._default_text_model) - await self.__get_vector_store(self._default_text_model).insert(vector, vedio.calculate_id()) - - async def __embedding_rich_text(self, rich_text: RichTextObject): - for document_id in rich_text.get_documents().values(): - document = DocumentObject.decode(self.env.get_knowledge_store().get_object_store().get_object(document_id)) - await self.__embedding_document(document) - for image_id in rich_text.get_images().values(): - image = ImageObject.decode(self.env.get_knowledge_store().get_object_store().get_object(image_id)) - await self.__embedding_image(image) - for video_id in rich_text.get_videos().values(): - video = VideoObject.decode(self.env.get_knowledge_store().get_object_store().get_object(video_id)) - await self.__embedding_video(video) - for rich_text_id in rich_text.get_rich_texts().values(): - rich_text = RichTextObject.decode(self.env.get_knowledge_store().get_object_store().get_object(rich_text_id)) - await self.__embedding_rich_text(rich_text) - - async def __embedding_email(self, email: EmailObject): - vector = await ComputeKernel.get_instance().do_text_embedding(json.dumps(email.get_desc()), self._default_text_model) - await self.__get_vector_store(self._default_text_model).insert(vector, email.calculate_id()) - await self.__embedding_rich_text(email.get_rich_text()) - - - async def __do_embedding(self, object: KnowledgeObject): - if object.get_object_type() == ObjectType.Document: - await self.__embedding_document(object) - if object.get_object_type() == ObjectType.Image: - await self.__embedding_image(object) - if object.get_object_type() == ObjectType.Video: - await self.__embedding_video(object) - if object.get_object_type() == ObjectType.RichText: - await self.__embedding_rich_text(object) - if object.get_object_type() == ObjectType.Email: - await self.__embedding_email(object) - else: - pass - - async def parse(self, object: ObjectID) -> str: - obj = self.env.get_knowledge_store().load_object(object) - await self.__do_embedding(obj) - return "insert into vector store" - -def init(env: KnowledgePipelineEnvironment, params: dict) -> EmbeddingParser: - return EmbeddingParser(env, params) \ No newline at end of file diff --git a/src/component/knowledge_manager/pipeline.py b/src/component/knowledge_manager/pipeline.py index 1701f4f..efe0936 100644 --- a/src/component/knowledge_manager/pipeline.py +++ b/src/component/knowledge_manager/pipeline.py @@ -23,10 +23,6 @@ class KnowledgePipelineManager: "names": {}, "running": [] } - from .input import local_dir - self.register_input("local_dir", local_dir.init) - from .parser import embedding - self.register_parser("embedding", embedding.init) def register_input(self, name: str, init_method): self.input_modules[name] = init_method @@ -84,6 +80,6 @@ class KnowledgePipelineManager: config = toml.load(f) for path in config["pipelines"]: pipeline_path = os.path.join(root, path) - with open(os.path.join(pipeline_path, "pipeline.toml")) as f: + with open(os.path.join(pipeline_path, "pipeline.toml"), 'r', encoding='utf-8') as f: pipeline_config = toml.load(f) self.add_pipeline(pipeline_config, pipeline_path) diff --git a/src/component/mail_environment/1/mail.json b/src/component/mail_environment/1/mail.json new file mode 100644 index 0000000..8a30f00 --- /dev/null +++ b/src/component/mail_environment/1/mail.json @@ -0,0 +1,6 @@ +{ + "subject": "开发dmc开源客户端", + "from_addr": "sichangjun@buckyos.com", + "to_addr": ["liuzhicong@buckyos.com"], + "date": "2023-4-10 21:00" +} \ No newline at end of file diff --git a/src/component/mail_environment/1/mail.txt b/src/component/mail_environment/1/mail.txt new file mode 100644 index 0000000..1abf50b --- /dev/null +++ b/src/component/mail_environment/1/mail.txt @@ -0,0 +1,51 @@ +最小功能集：4.15准备好oktc合约和客户端工具；4.21之前完成一些矿场节点部署； +Oktc合约和rust接口 （秋总） +实现以DMC结算的bill 和 order （done） +实现链上挑战—— merkle联通证明；用户提交低深度半路径；矿工提供叶子原文和高深度半路径；（doing） +实现按照价格和质押率索引的spv节点 —— 支持用户按参数匹配下单；（doing） +实现对eth发起的http请求 auth 头认证 —— 支持https实现 链下的 write/restore/challenge 身份认证；（doing） +实现oktc client event listener的block number本地持久化；（doing） +用户端工具：面向普通存储用户，一键备份和恢复； +源数据管理服务： +注册本地路径，生成链下挑战密码本 ——随机偏移和长度QA（done） + 生成链上挑战密码本 —— merkle根和半深度茎节点（doing）； +账号服务： +添加本地路径，选择bill id发起order（done） +按参数匹配order ——依赖按参数索引的spv 服务（doing） +生成order提交到交付服务；（done） +交付服务 +注册order和关联的源数据（done） +提交merkle根（done） +监听链事件，同步order状态，向miner写入源数据；（done） +使用密码本持续发起链下挑战（done） +实现链上挑战（doing） +恢复数据到本地目录（done） +关键日志服务 +关键状态改变日志写入（doing） +关键状态日志事件监听接口（doing） +轻量命令行客户端 +服务部署脚本（doing） +传入本地文件 和 order参数一键完成备份 （doing） +一键恢复（doing） +链下挑战失效时，自动发起链上挑战 ——依赖关键日志服务（doing） +mysql实现移植到sqlite（看时间和问题多不多；undo） +矿场端工具：本版本不是发布重点；保证能在自由的节点上稳定运行即可；结构上保证了一定的伸缩性； +存储扇区管理服务：包括gateway 和 node；保证简单部署和扩展，可靠性暂时不需实现； +扇区gateway服务：注册node，分发扇区读写请求到node；（doing） +扇区node服务：注册本地目录到node，注册为可用扇区；（测试中单node可以直接接入，done） +账号服务： +添加扇区和挂单参数，生成bill（done）； +监听链事件，响应新的order转入交付服务；（done） +监听链事件，响应链上挑战转入交付服务；（doing） +交付服务： +注册order和关联的扇区；（done） +对user提供交付相关http接口： +查询miner端order状态（done） +写入源数据（done） +完成写入后准备生成merkle root准备证明（done） +恢复源数据（done） +接收链下挑战返回证明（done） +自动提交链上挑战（doing） +关键日志服务 +关键状态改变日志写入（可推迟实现；undo） +自有节点部署 \ No newline at end of file diff --git a/src/component/mail_environment/__init__.py b/src/component/mail_environment/__init__.py new file mode 100644 index 0000000..402cf90 --- /dev/null +++ b/src/component/mail_environment/__init__.py @@ -0,0 +1,3 @@ +from .issue import IssueParser +from .local import LocalEmail +from .spider import EmailSpider \ No newline at end of file diff --git a/src/component/mail_environment/issue.py b/src/component/mail_environment/issue.py new file mode 100644 index 0000000..53ed3c6 --- /dev/null +++ b/src/component/mail_environment/issue.py @@ -0,0 +1,200 @@ +# define a knowledge base class +import json +import string +from aios_kernel import ComputeKernel, AIStorage, Environment, SimpleAIFunction, BaseAIAgent, AgentPrompt, AgentMsg +from knowledge import * +from .mail import MailStorage, Mail + +class IssueState(Enum): + Open = 1 + InProgress = 2 + Closed = 3 + +class IssueUpdateHistory: + def __init__(self, source: str, changes: dict) -> None: + self.source = source + self.changes = changes + + +class Issue: + def __init__(self) -> None: + self.id = None + self.summary = "" + self.state = IssueState.Open + self.source: str = None + self.create_time: datetime = None + self.deadline: datetime = None + self.update_history = [] + self.children = [] + self.parent: ObjectID = None + + @classmethod + def object_type(cls) -> ObjectType: + return ObjectType.from_user_def_type_code(0) + + def to_prompt(self) -> str: + prompt = { + "id": self.id, + "summary": self.summary, + "state": self.state.name, + "deadline": self.deadline + } + return json.dumps(prompt) + + @classmethod + def prompt_desc(cls) -> str: + return '''a issue contains following fileds: { + id: a guid string to identify a issue + summary: summary of this issue + state: state of this issue, will be one of [Open, InProgress, Closed], + deadline: if issue is not closed, deadline is the time to close this issue + } + ''' + + def calculate_id(self) -> str: + desc = { + "summary": self.summary, + "source": self.source, + "create_time": self.create_time, + "deadline": self.deadline, + "parent": self.parent, + } + id = str(KnowledgeObject(Issue.object_type(), desc).calculate_id()) + self.id = id + return id + + +class IssueStorage: + def __init__(self, path: str, root: Issue=None) -> None: + self.path = path + if not os.path.exists(path): + self.root = root + else: + self.root = json.load(open(path, "r")) + + def __flush(self): + json.dump(self.root, open(self.path, "w")) + + def get_issue_by_id(self, id: str) -> Issue: + self.root() + + def __get_issue_by_mail_in_subtree(self, root_issue: Issue, mail_id: str): + if root_issue.source == mail_id: + return root_issue + if root_issue.children is None or len(root_issue.children) == 0: + return None + for child_issue in root_issue.children: + this_issue = self.__get_issue_by_mail_in_subtree(child_issue, mail_id) + if this_issue is not None: + return this_issue + return None + + def get_issue_by_mail(self, mail_storage: MailStorage, mail: Mail) -> Issue: + if mail.reply_to is None: + return self.root + this_mail = mail_storage.get_mail_by_id(mail.reply_to) + while True: + issue = self.__get_issue_by_mail_in_subtree(self.root, this_mail.id) + if issue is not None: + return issue + if this_mail.replay_to is None: + return self.root + this_mail = mail_storage.get_mail_by_id(this_mail.reply_to) + + + def add_issue(self, source_id: str, issue: dict): + parent_id = issue.get("parent") + parent_issue = self.get_issue(parent_id) + issue: Issue = issue + issue["source"] = source_id + issue.calculate_id() + parent_issue.children.append(issue) + self.__flush() + + def update_issue(self, source_id: str, update: dict): + issue = self.get_issue(update["id"]) + source = update["source"] + changes = {} + for key, value in update.items(): + if key != "id" and key is not "source": + changes[key] = { + "old": issue[key], + "new": value, + } + issue[key] = value + issue.update_history.append(IssueUpdateHistory(source, changes)) + + self.__flush() + + +class IssueParserEnvironment(Environment): + def __init__(self, env_id: str, storage: IssueStorage) -> None: + super().__init__(env_id) + self.storage = storage + + update_description = '''update issue with email object''' + + update_param = { + "source_id": "update issue with which email object id", + "update_content": '''issue fileds to update, json format; + if id field exists, update the issue with the id; + if id filed not exists, create a new issue with the content; + other fileds in update_content will be updated to the issue; + ''', + } + self.add_ai_function(SimpleAIFunction("update_issue", + update_description, + self._update, + update_param)) + + async def _update(self, source_id: str, update_content: str): + update_issue = json.loads(update_content) + issue_id = update_issue.get("id") + if issue_id: + self.storage.update_issue(source_id, update_issue) + else: + self.storage.add_issue(source_id, update_issue) + + + +class IssueParser: + def __init__(self, env: KnowledgePipelineEnvironment, config: dict): + mail_path = string.Template(config["mail_path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir()) + issue_path = string.Template(config["issue_path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir()) + config["path"] = issue_path + + self.env = env + self.config = config + self.mail_storage = MailStorage(mail_path) + + root_issue = None + if "root_issue" in config: + root_issue = Issue() + root_issue.summary = config["root_issue"] + self.issue_storage = IssueStorage(issue_path, root_issue) + self.llm_env = IssueParserEnvironment("issue_parser", self.issue_storage) + + def get_path(self) -> str: + return self.config["path"] + + async def parse(self, mail_id: ObjectID) -> str: + mail_id = str(mail_id) + mail = self.mail_storage.get_mail_by_id(mail_id) + issue = self.issue_storage.get_issue_by_mail(self.mail_storage, mail) + mail_str = mail.to_prompt() + issue_str = issue.to_prompt() + + mail_desc = Mail.prompt_desc() + issue_desc = Issue.prompt_desc() + prompt = f'''I'll give a mail in json format, {mail_desc}; + and a issue in json format, {issue_desc}; + you should read this mail {mail_str}, see if this mail associated with this issue {issue_str}; + if this mail is about a new child issue of this issue, create a new issue with this mail, fill param update_content's summary field will mail content, set parent field with id of this issue; + if this mail will update this issue, set id filed to this issue, fill update_content param with new summary and new state with this mail content; + then you should call update_issue function with source_id set to this mail id, and update_content in json format; + if this mail is not associated with issue, you should ignore this mail without an function call; + ''' + + llm_result = await BaseAIAgent.do_llm_complection(self.llm_env, AgentPrompt(prompt), AgentMsg(), "gpt-4", 16000) + return "update issue" + diff --git a/src/component/mail_environment/local.py b/src/component/mail_environment/local.py new file mode 100644 index 0000000..07b038d --- /dev/null +++ b/src/component/mail_environment/local.py @@ -0,0 +1,34 @@ +import os +import logging +import json +import string +from knowledge import * +from aios_kernel.storage import AIStorage +from .mail import Mail, MailStorage + + +class LocalEmail: + def __init__(self, env: KnowledgePipelineEnvironment, config:dict): + self.config = config + self.env = env + path = string.Template(config["path"]).substitute(myai_dir=AIStorage.get_instance().get_myai_dir()) + self.mail_storage = MailStorage(path, config.get("watch")) + + async def next(self): + while True: + parsed = None + journals = self.env.journal.latest_journals(1) + if len(journals) == 1: + latest_journal = journals[0] + if latest_journal.is_finish(): + yield None + continue + parsed = str(latest_journal.get_object_id()) + + mail_id = self.mail_storage.next_mail_id(parsed) + if mail_id is None: + yield (None, None) + else: + yield (mail_id, str(mail_id)) + + \ No newline at end of file diff --git a/src/component/mail_environment/mail.py b/src/component/mail_environment/mail.py new file mode 100644 index 0000000..ff5ef15 --- /dev/null +++ b/src/component/mail_environment/mail.py @@ -0,0 +1,265 @@ +import asyncio +import json +import mailparser +import base64 +import requests +import datetime +from bs4 import BeautifulSoup +import sqlite3 +import html2text +from knowledge import * + +class Mail: + def __init__(self, **kwargs) -> None: + self.from_addr = kwargs.get("From") + self.to_addr = kwargs.get("To") + self.subject = kwargs.get("Subject") + self.date = kwargs.get("Date") + self.bcc = kwargs.get("BCC") + self.cc = kwargs.get("CC") + self.reply_to = None + self.id: str = None + self.content: str = None + + def to_prompt(self) -> str: + prompt = { + "id": self.id, + "subject": self.subject, + "from": self.from_addr, + "date": self.date, + "content": self.content + } + return json.dumps(prompt) + + @classmethod + def prompt_desc(cls) -> dict: + return '''a mail contains following fileds: { + id: a guid string to identify a mail + subject: subject of this mail + from: sender address of this mail + date: date of this mail + content: content of this mail + } + ''' + + def get_date(self) -> datetime.datetime: + datetime.datetime.strptime(self.date, "%Y-%m-%d %H:%M") + + def calculate_id(self) -> str: + desc = { + "from_addr": self.from_addr, + "to_addr": self.to_addr, + "subject": self.subject, + "date": self.date, + "content": self.content, + "reply_to": self.reply_to + } + id = str(KnowledgeObject(ObjectType.Email, desc).calculate_id()) + self.id = id + return id + +class MailStorage: + def __init__(self, root, watch=False): + self.root = root + if not os.path.exists(root): + os.makedirs(root) + db_file = os.path.join(root, "mail.db") + + self.conn = sqlite3.connect(db_file) + cursor = self.conn.cursor() + cursor.execute( + """ + CREATE TABLE IF NOT EXISTS mails ( + uid INTEGER PRIMARY KEY, + object_id TEXT, + date DATETIME, + from_addr TEXT + ) + """ + ) + + if watch: + asyncio.create_task(self.watch_root()) + + def object_id_to_uid(self, object_id): + cursor = self.conn.cursor() + cursor.execute( + """ + SELECT uid FROM mails WHERE object_id = ? + """, + (object_id,), + ) + row = cursor.fetchone() + if row: + return row[0] + return None + + def uid_to_object_id(self, uid): + cursor = self.conn.cursor() + cursor.execute( + """ + SELECT object_id FROM mails WHERE uid = ? + """, + (uid,), + ) + row = cursor.fetchone() + if row: + return row[0] + return None + + def lastest_uid(self): + cursor = self.conn.cursor() + cursor.execute( + """ + SELECT uid FROM mails ORDER BY uid DESC LIMIT 1 + """ + ) + row = cursor.fetchone() + if row: + return row[0] + return None + + def lastest_mail_id(self): + cursor = self.conn.cursor() + cursor.execute( + """ + SELECT object_id FROM mails ORDER BY uid DESC LIMIT 1 + """ + ) + row = cursor.fetchone() + if row: + return row[0] + return None + + def next_mail_id(self, id): + uid = 0 if id is None else self.object_id_to_uid(id) + cursor = self.conn.cursor() + cursor.execute( + """ + SELECT object_id FROM mails WHERE uid > ? ORDER BY uid ASC LIMIT 1 + """, + (uid,), + ) + row = cursor.fetchone() + if row: + return row[0] + return None + + + + def get_mail_by_id(self, id): + uid = self.object_id_to_uid(id) + mail = Mail() + mail.id = id + mail_dir = self.mail_dir(uid) + mail_json = json.load(open(f"{mail_dir}/mail.json", "r", encoding='utf-8')) + mail.__dict__.update(mail_json) + with open(f"{mail_dir}/mail.txt", "r", encoding='utf-8') as f: + mail_content = f.read() + mail.content = mail_content + return mail + + def mail_dir(self, uid): + return os.path.join(self.root, str(uid)) + + # for debug + async def watch_root(self): + while True: + latest_uid = self.lastest_uid() + for uid in os.listdir(self.root): + mail_dir = os.path.join(self.root, uid) + if uid.isdigit() and os.path.isdir(mail_dir): + uid = int(uid) + if uid <= latest_uid: + continue + mail = Mail() + mail_json = json.load(open(f"{mail_dir}/mail.json", "r", encoding='utf-8')) + + mail.__dict__.update(mail_json) + # mail content + with open(f"{mail_dir}/mail.txt", "r", encoding='utf-8') as f: + mail_content = f.read() + mail.content = mail_content + mail.calculate_id() + cursor = self.conn.cursor() + cursor.execute( + """ + INSERT INTO mails (uid, object_id, date, from_addr) + VALUES (?, ?, ?, ?) + """, + (uid, mail.id, mail.get_date(), mail.from_addr), + ) + self.conn.commit() + await asyncio.sleep(10) + + def download(self, uid, mail: mailparser.MailParser): + mail_dir = self.mail_dir(uid) + os.makedirs(dir) + + meta = json.loads(mail.mail_json) + mail = Mail(**meta) + reply_to = meta.get("In-Reply-To") + if reply_to: + mail.reply_to = self.uid_to_object_id(reply_to) + + h = html2text.HTML2Text() + h.ignore_links = True + h.ignore_images = True + mail_content = h.handle(mail.body) + mail.content = mail_content + + mail.calculate_id() + del mail.content + json.dump(mail.__dict__, open(f"{mail_dir}/mail.json", "w", encoding='utf-8')) + + # save mail content + with open(f"{mail_dir}/mail.txt", "w", encoding='utf-8') as f: + f.write(mail_content) + + for attachment in mail.attachments: + if attachment['mail_content_type'] in ['image/png', 'image/jpeg', 'image/gif']: + filename = attachment['filename'] + filefullname = f"{mail_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") + + # get all image urls + soup = BeautifulSoup(mail.body, '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 + + 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(mail_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}') + + cursor = self.conn.cursor() + cursor.execute( + """ + INSERT INTO mails (uid, object_id, date, from_addr) + VALUES (?, ?, ?, ?) + """, + (uid, mail.id, mail.date, mail.from_addr), + ) + + + \ No newline at end of file diff --git a/src/component/mail_environment/spider.py b/src/component/mail_environment/spider.py new file mode 100644 index 0000000..2502468 --- /dev/null +++ b/src/component/mail_environment/spider.py @@ -0,0 +1,53 @@ +import os +import logging +import json +import imaplib +import mailparser +from knowledge import * +from aios_kernel.storage import AIStorage + + +class EmailSpider: + def __init__(self, env: KnowledgePipelineEnvironment, config:dict): + self.config = config + self.env = env + self.env.get_logger().info(f"read email config from {self.config.get('imap_server')}") + self.client = imaplib.IMAP4_SSL( + host=self.config.get('imap_server'), + port=self.config.get('imap_port') + ) + self.client.login(self.config.get('address'), self.config.get('password')) + self.mail_local_root = os.path.join(self.env.pipeline_path, self.config.get("address")) + os.makedirs(self.mail_local_root) + + async def next(self): + while True: + _, data = self.client.uid('search', None, "ALL") + uid_list = data[0].split() + if uid_list.len() == 0: + yield (None, None) + continue + + journals = self.env.journal.latest_journals(1) + from_uid = 0 + if len(journals) == 1: + latest_journal = journals[0] + if latest_journal.is_finish(): + yield None + continue + from_uid = int(latest_journal.get_input()) + if int.from_bytes(uid_list[-1]) <= from_uid: + yield (None, None) + continue + + for uid in uid_list: + _uid = int.from_bytes(uid) + if _uid > from_uid: + message_parts = "(BODY.PEEK[])" + _, email_data = self.client.uid('fetch', uid, message_parts) + mail = mailparser.parse_from_bytes(email_data[0][1]) + self.save_email(_uid, mail) + + yield (None, None) + + \ No newline at end of file diff --git a/src/knowledge/object/object.py b/src/knowledge/object/object.py index 75852c3..f34da2d 100644 --- a/src/knowledge/object/object.py +++ b/src/knowledge/object/object.py @@ -51,13 +51,13 @@ class KnowledgeObject(ABC): def get_summary(self) -> str: return self.desc.get("summary") - def get_articl_catelog(self) -> str: - assert self.object_type == ObjectType.Document - return self.desc.get("catelog") + # def get_articl_catelog(self) -> str: + # assert self.object_type == ObjectType.Document + # return self.desc.get("catelog") - def get_article_full_content(self) -> str: - assert self.object_type == ObjectType.Document - return self.body + # def get_article_full_content(self) -> str: + # assert self.object_type == ObjectType.Document + # return self.body def calculate_id(self): # Convert the object_type and desc to string and compute the SHA256 hash @@ -73,6 +73,6 @@ class KnowledgeObject(ABC): def encode(self) -> bytes: return pickle.dumps(self) - @staticmethod - def decode(data: bytes) -> "ImageObject": - return pickle.loads(data) + # @staticmethod + # def decode(data: bytes) -> "ImageObject": + # return pickle.loads(data) diff --git a/src/knowledge/object/object_id.py b/src/knowledge/object/object_id.py index 46955f4..3a07b2b 100644 --- a/src/knowledge/object/object_id.py +++ b/src/knowledge/object/object_id.py @@ -13,6 +13,17 @@ class ObjectType(IntEnum): Document = 103 RichText = 104 Email = 105 + UserDef = 200 + + def is_user_def(self) -> bool: + return self.value >= 200 + + def get_user_def_type_code(self): + return (self.value - 200) if self.is_user_def() else None + + @classmethod + def from_user_def_type_code(value): + return value + 200 # define a object ID class to identify a object diff --git a/src/knowledge/pipeline.py b/src/knowledge/pipeline.py index 62f246a..b902225 100644 --- a/src/knowledge/pipeline.py +++ b/src/knowledge/pipeline.py @@ -14,6 +14,9 @@ class KnowledgePipelineJournal: def is_finish(self) -> bool: return self.object_id is None + + def get_object_id(self) -> ObjectID: + return self.object_id def get_input(self) -> str: return self.input diff --git a/src/service/aios_shell/aios_shell.py b/src/service/aios_shell/aios_shell.py index 11c5036..ccbce5a 100644 --- a/src/service/aios_shell/aios_shell.py +++ b/src/service/aios_shell/aios_shell.py @@ -45,7 +45,6 @@ shell_style = Style.from_dict({ 'error': '#8F0000 bold' }) - class AIOS_Shell: def __init__(self,username:str) -> None: self.username = username