【Python+微信】【企业微信开发入坑指北】4. 企业微信接入GPT，只需一个URL，自动获取文章总结

if decrypt_data.get('Content', '').find("http://") != -1 or decrypt_data.get('Content', '').find("https://") != -1:
        response_content = await summary_url(decrypt_data.get('Content', ''))

Messages = {}
msg_id = decrypt_data.get('MsgId')
if (msg_id in Messages.keys()):
    return "success"
Messages[msg_id] = decrypt_data

if decrypt_data.get('Content', '').find("http://") != -1 or decrypt_data.get('Content', '').find("https://") != -1:
    response_content = await summary_url(decrypt_data.get('Content', ''))
    weichat_app.send_text(content=response_content)

def get_outline(page):
    soup = _get_soup(page.html)
    outline = []
    def process_element(element, depth):
        name = element.name
        if not name:
            return
        if name in ["script", "style"]:
            return
        element_info = {"name": element.name, "depth": depth}
        if name in ["svg"]:
            element_info["text"] = None
            outline.append(element_info)
            return
        element_info["text"] = element.string
        # Check if the element has an "id" attribute
        if "id" in element.attrs:
            element_info["id"] = element["id"]
        if "class" in element.attrs:
            element_info["class"] = element["class"]
        outline.append(element_info)
        for child in element.children:
            process_element(child, depth + 1)
    try:
        for element in soup.body.children:
            process_element(element, 1)
    except:
        logger.error("get outline error")
        outline = []
    return outline
PROMPT_TEMPLATE = """Please complete the web page crawler parse function to achieve the User Requirement. The parse \
function should take a BeautifulSoup object as input, which corresponds to the HTML outline provided in the Context.
``python
from bs4 import BeautifulSoup
# only complete the parse function
def parse(soup: BeautifulSoup):
    ...
    # Return the object that the user wants to retrieve, don't use print
``
## User Requirement
{requirement}
## Context
The outline of html page to scrabe is show like below:
``tree
{outline}
``
"""
class WriteCrawlerCode(Action):
    async def run(self, url, requirement):
        codes = {}
        codes[url] = await self._write_code(url, requirement)
        if codes[url] == None:
            return None
        return "\n".join(f"# {url}\n{code}" for url, code in codes.items()) ## 返回固定格式的url + 相应爬虫代码
    async def _write_code(self, url, query):
        page = await WebBrowserEngine().run(url)
        outline = get_outline(page)
        if len(outline) == 0:
            return None
        outline = "\n".join(
            f"{' '*i['depth']}{'.'.join([i['name'], *i.get('class', [])])}: {i['text'] if i['text'] else ''}"
            for i in outline
        )
        code_rsp = await self._aask(PROMPT_TEMPLATE.format(outline=outline, requirement=query))
        code = CodeParser.parse_code(block="", text=code_rsp)
        return code

class RunCrawlerCode(Action):
    async def run(self, url, codes):
        code, current = codes.rsplit(f"# {url}", maxsplit=1)
        name = uuid4().hex
        module = type(sys)(name)
        exec(current, module.__dict__)
        page = await WebBrowserEngine().run(url)
        data = getattr(module, "parse")(page.soup)
        print(data)
        return str(data)  # 以字符串形式返回

from metagpt.roles import Role
class CrawlerEngineer(Role):
    ......
    def __init__(self, **kwargs) -> None:
        super().__init__(**kwargs)
        self.set_actions([WriteCrawlerCode, RunCrawlerCode])
        
    async def _think(self) -> None:
        ......
    async def _act(self) -> Message:
        """Perform an action as determined by the role.
        Returns:
            A message containing the result of the action.
        """
        todo = self.rc.todo
        if type(todo) is WriteCrawlerCode:
            resp = await todo.run(url=self.url, requirement=self.requirement)
            logger.info(resp)
            if (resp == None):
                return None
            self.rc.memory.add(Message(content=resp, role=self.profile))
            return resp
        msg = self.rc.memory.get(k=1)[0]
        resp = await todo.run(url=self.url, codes=msg.content) # 返回必须是字符串
        logger.info(resp)
        return Message(content=resp, role=self.profile) # resp必须是字符串，MetaGPT中限制的
    async def _react(self) -> Message:
        ......

class CrawlerDataProvider():
    def __init__(self) -> None:
        pass
    
    async def run(self, url, requirement="获取正文中的所有文字内容，如果正文有code，将code也作为文字内容"):
        msg = "start"
        role = CrawlerEngineer(url = url, requirement = requirement)
        logger.info(msg)
        result = await role.run(msg)
        logger.info("\n=========================================\n")
        logger.info(result)
        return result

if __name__ == "__main__":
    url="https://mp.weixin.qq.com/s/2m8MrsCxf5boiH4Dzpphrg"
    requirement="获取标题，正文中的所有文字内容，如果正文有code，将code也作为文字内容"
    data_provider = CrawlerDataProvider()
    asyncio.run(data_provider.run(url, requirement))

PROMPT_TEMPLATE = """简要总结下面文字的内容:
"{text}"
简要总结:"""
class SummaryArticle():
    def __init__(self) -> None:
        pass
    
    def run(self, text):
        prompt = PROMPT_TEMPLATE.format(text = text)
        response_content = openai_wrapper.get_chat_completion(prompt)
        print("response content: ", response_content)
        return response_content

async def summary_url(url):
    text_provider = CrawlerDataProvider()
    text = await text_provider.run(url = url)
    summary = SummaryArticle()
    response_content = summary.run(text = text)
    return response_content
if __name__ == "__main__":
    response = asyncio.run(summary_url(url = "https://mp.weixin.qq.com/s/L_gHW-_TIipmcyDcdQpZRA"))
    print(response)

pip install bs4