Hanah

# どの役割でもインポート可能で、初期化し、開始メッセージで実行し、完了！
from metagpt.roles.product_manager import ProductManager
prompt = f"""
# 役割：ソフトウェア開発チーム

## 背景 :

私はソフトウェア開発チームです。
今、html、js、vue3、element-plusを使用して問題解決プログラムを開発する必要があります。
問題解決は、人々が問題に関連する知識をより深く理解するのに役立ちます。

## プロフィール:
- 著者: 黎伟
- バージョン: 0.1
- 言語: 中国語
- 説明: 私はソフトウェア開発チームです。

## 目標:
- html、js、vue3、element-plusを使用して問題解決プログラムの開発要件文書を作成します。

## 制約:
1. 最終的に納品されるプログラムはhtml単一ファイルであり、他のファイルは不要です。
2. 問題の形式は、少なくとも2つの判断問題、2つの選択問題、2つの空欄問題を含む必要があります。
3. 問題の内容は、人工知能のエージェントの基本理論に関連している必要があります。
4. 問題解決プログラムは、少なくとも10のサンプル問題を提供する必要があります。
5. 問題は、htmlファイルのscript部分にリスト形式で記述される必要があります。
6. vue3、element-plusはcdn形式でhtmlのheader部分にインポートされる必要があります。

## スキル:
1. 強力なjs言語開発能力を持つ
2. vue3、element-plusの使用に精通している
3. 人工知能のエージェントの基本理論を良く理解している
4. タイポグラフィの美学を持ち、番号、インデント、区切り線、改行などを利用して情報のレイアウトを美化することができる


上記の要件に基づいて、問題解決プログラムの開発要件文書を完成させてください。
"""
async def main():
    role = ProductManager()
    result = await role.run(prompt)
    
await main()

from metagpt.actions import Action

class SimpleWriteCode(Action):
    PROMPT_TEMPLATE: str = """
    Write a python function that can {instruction} and provide two runnnable test cases.
    Return ```python your_code_here ```with NO other texts,
    your code:
    """

    name: str = "SimpleWriteCode"

    async def run(self, instruction: str):
        prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)

        rsp = await self._aask(prompt)

        code_text = SimpleWriteCode.parse_code(rsp)

        return code_text

    @staticmethod
    def parse_code(rsp):
        pattern = r"```python(.*)```"
        match = re.search(pattern, rsp, re.DOTALL)
        code_text = match.group(1) if match else rsp
        return code_text

import re
import os
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import logger

class SimpleCoder(Role):
    name: str = "Alice"
    profile: str = "SimpleCoder"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([SimpleWriteCode])

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        todo = self.rc.todo  # todo will be SimpleWriteCode()

        msg = self.get_memories(k=1)[0]  # find the most recent messages
        code_text = await todo.run(msg.content)
        msg = Message(content=code_text, role=self.profile, cause_by=type(todo))

        return msg

async def main():
    msg = "write a function that calculates the sum of a list"
    role = SimpleCoder()
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)
    return result
    
        
rtn = await main()

2025-01-26 18:14:18.317 | INFO     | __main__:main:4 - write a function that calculates the sum of a list
2025-01-26 18:14:18.319 | INFO     | __main__:_act:16 - Alice(SimpleCoder): to do SimpleWriteCode(SimpleWriteCode)
```python
def sum_list(numbers):
    return sum(numbers)

# テストケース
test_case_1 = [1, 2, 3, 4, 5]
test_case_2 = [10, 20, 30, 40, 50]

# テストケース1
print(sum_list(test_case_1))  # 期待される出力: 15

# テストケース2
print(sum_list(test_case_2))  # 期待される出力: 150

2025-01-26 18:14:25.372 | WARNING  | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 18:14:25.374 | INFO     | __main__:main:6 - SimpleCoder: 
def sum_list(numbers):
    return sum(numbers)

# テストケース
test_case_1 = [1, 2, 3, 4, 5]
test_case_2 = [10, 20, 30, 40, 50]

# テストケース1
print(sum_list(test_case_1))  # 期待される出力: 15

# テストケース2
print(sum_list(test_case_2))  # 期待される出力: 150

# SimpleWriteCode このクラスは前のセクションと全く同じです

from metagpt.actions import Action

class SimpleWriteCode(Action):
    PROMPT_TEMPLATE: str = """
    Write a python function that can {instruction} and provide two runnnable test cases.
    Return ```python your_code_here ```with NO other texts,
    your code:
    """

    name: str = "SimpleWriteCode"

    async def run(self, instruction: str):
        prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)

        rsp = await self._aask(prompt)

        code_text = SimpleWriteCode.parse_code(rsp)

        return code_text

    @staticmethod
    def parse_code(rsp):
        pattern = r"```python(.*)```"
        match = re.search(pattern, rsp, re.DOTALL)
        code_text = match.group(1) if match else rsp
        return code_text

# 本節ではSimpleRunCodeクラスを追加します
class SimpleRunCode(Action):
    name: str = "SimpleRunCode"

    async def run(self, code_text: str):
        result = subprocess.run(["python", "-c", code_text], capture_output=True, text=True)
        code_result = result.stdout
        logger.info(f"{code_result=}")
        return code_result

import re
import os
import subprocess
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import logger
class RunnableCoder(Role):
    name: str = "Alice"
    profile: str = "RunnableCoder"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([SimpleWriteCode, SimpleRunCode])
        self._set_react_mode(react_mode="by_order")

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        # By choosing the Action by order under the hood
        # todo will be first SimpleWriteCode() then SimpleRunCode()
        todo = self.rc.todo

        msg = self.get_memories(k=1)[0]  # find the most k recent messages
        result = await todo.run(msg.content)

        msg = Message(content=result, role=self.profile, cause_by=type(todo))
        self.rc.memory.add(msg)
        return msg

async def main():
    msg = "write a function that calculates the sum of a list"
    role = RunnableCoder()
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)
    return result

rtn = await main()
print(rtn)

def calculate_sum(numbers):
    return sum(numbers)

# テストケース
def test_case_1():
    assert calculate_sum([1, 2, 3, 4, 5]) == 15

def test_case_2():
    assert calculate_sum([10, -5, 0, 7, 3]) == 15

from metagpt.actions import Action
from typing import Dict, Union
import ast

def extract_struct(text: str, data_type: Union[type(list), type(dict)]) -> Union[list, dict]:
    """指定されたタイプの構造（辞書またはリスト）を与えられたテキストから抽出し、解析します。
    テキストには、ネストされた構造を持つリストまたは辞書のみが含まれています。

    Args:
        text: 構造を含むテキスト（辞書またはリスト）。
        data_type: 抽出するデータタイプ、"list"または"dict"。

    Returns:
        - 抽出と解析が成功した場合、対応するデータ構造（リストまたは辞書）を返します。
        - 抽出が失敗した場合や解析中にエラーが発生した場合は、例外をスローします。
    例:
        >>> text = 'xxx [1, 2, ["a", "b", [3, 4]], {"x": 5, "y": [6, 7]}] xxx'
        >>> result_list = OutputParser.extract_struct(text, "list")
        >>> print(result_list)
        >>> # 出力: [1, 2, ["a", "b", [3, 4]], {"x": 5, "y": [6, 7]}]

        >>> text = 'xxx {"x": 1, "y": {"a": 2, "b": {"c": 3}}} xxx'
        >>> result_dict = OutputParser.extract_struct(text, "dict")
        >>> print(result_dict)
        >>> # 出力: {"x": 1, "y": {"a": 2, "b": {"c": 3}}}
    """
    # 最初の "[" または "{" と最後の "]" または "}" を見つける
    start_index = text.find("[" if data_type is list else "{")
    end_index = text.rfind("]" if data_type is list else "}")

    if start_index != -1 and end_index != -1:
        # 構造部分を抽出する
        structure_text = text[start_index : end_index + 1]

        try:
            # ast.literal_evalを使用してテキストをPythonデータ型に変換しようとする
            result = ast.literal_eval(structure_text)

            # 結果が指定されたデータタイプと一致することを確認する
            if isinstance(result, list) or isinstance(result, dict):
                return result

            raise ValueError(f"抽出された構造は{data_type}ではありません。")

        except (ValueError, SyntaxError) as e:
            raise Exception(f"{data_type}の抽出と解析中にエラーが発生しました: {e}")
    else:
        logger.error(f"テキストに{data_type}が見つかりません。")
        return [] if data_type is list else {}

class WriteDirectory(Action):
    """チュートリアルディレクトリを書くためのアクションクラス。

    Args:
        name: アクションの名前。
        language: 出力する言語、デフォルトは"Chinese"。
        
        チュートリアルディレクトリを書くためのアクションクラス。
        引数：
        name：アクションの名前。
        language：出力する言語、デフォルトは"Chinese"。
    """

    name: str = "WriteDirectory"
    language: str = "Chinese"

    async def run(self, topic: str, *args, **kwargs) -> Dict:
        """トピックに従ってチュートリアルディレクトリを生成するアクションを実行します。

        Args:
            topic: チュートリアルのトピック。

        Returns:
            チュートリアルディレクトリ情報、{"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}を含む。
        トピックに従ってチュートリアルディレクトリを生成するアクションを実行します。
            引数：
            topic：チュートリアルのトピック。
            戻り値：
            チュートリアルディレクトリ情報、{"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}を含む。
        """
        COMMON_PROMPT = """
        あなたは今、インターネットの分野で経験豊富な技術専門家です。 
        トピック"{topic}"に関する技術チュートリアルを書く必要があります。
        """

        DIRECTORY_PROMPT = COMMON_PROMPT + """
        このチュートリアルの具体的な目次を提供してください。以下の要件に厳密に従ってください：
        1. 出力は指定された言語、{language}に厳密に従う必要があります。
        2. 回答は厳密に辞書形式で、{{"title": "xxx", "directory": [{{"dir 1": ["sub dir 1", "sub dir 2"]}}, {{"dir 2": ["sub dir 3", "sub dir 4"]}}]}}のようにしてください。
        3. 目次はできるだけ具体的かつ十分で、一次および二次の目次を含む必要があります。二次の目次は配列にします。
        4. 余分なスペースや改行はありません。
        5. 各目次タイトルには実際の意味があります。
        """
        prompt = DIRECTORY_PROMPT.format(topic=topic, language=self.language)
        resp = await self._aask(prompt=prompt)
        return extract_struct(resp, dict)

text = 'xxx [1, 2, ["a", "b", [3, 4]], {"x": 5, "y": [6, 7]}] xxx'
result_list = extract_struct(text, list)
print(result_list)
# 出力: [1, 2, ["a", "b", [3, 4]], {"x": 5, "y": [6, 7]}]

text = 'xxx {"x": 1, "y": {"a": 2, "b": {"c": 3}}} xxx'
result_dict = extract_struct(text, dict)
print(result_dict)
# 出力: {"x": 1, "y": {"a": 2, "b": {"c": 3}}}

from metagpt.const import TUTORIAL_PATH
TUTORIAL_PATH
# WindowsPath('d:/Code/agent_from_scratch/data/tutorial_docx')

from datetime import datetime
from typing import Dict
import asyncio
from metagpt.actions.write_tutorial import WriteDirectory, WriteContent
from metagpt.const import TUTORIAL_PATH
from metagpt.logs import logger
from metagpt.roles.role import Role, RoleReactMode
from metagpt.schema import Message
from metagpt.utils.file import File

from typing import Dict

from metagpt.actions import Action

class WriteDirectory(Action):
    """チュートリアルディレクトリを書くためのアクションクラス。

    Args:
        name: アクションの名前。
        language: 出力する言語、デフォルトは"Chinese"。
    """

    name: str = "WriteDirectory"
    language: str = "Chinese"

    async def run(self, topic: str, *args, **kwargs) -> Dict:
        """トピックに従ってチュートリアルディレクトリを生成するアクションを実行します。

        Args:
            topic: チュートリアルのトピック。

        Returns:
            チュートリアルディレクトリ情報、{"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}を含む。
        """
        COMMON_PROMPT = """
        あなたは今、インターネットの分野で経験豊富な技術専門家です。 
        トピック"{topic}"に関する技術チュートリアルを書く必要があります。
        """
        CONTENT_PROMPT = COMMON_PROMPT + """
        今、あなたにモジュールディレクトリタイトルを与えます。 
        このタイトルの詳細な原則内容を詳しく出力してください。 
        コード例がある場合は、標準のコード仕様に従って提供してください。 
        コード例がない場合は、必要ありません。

        トピックのモジュールディレクトリタイトルは次のとおりです：
        {directory}

        出力を次の要件に厳密に制限してください：
        1. レイアウトのMarkdown構文形式に従ってください。
        2. コード例がある場合は、標準の構文仕様に従い、ドキュメント注釈を持ち、コードブロックに表示される必要があります。
        3. 出力は指定された言語、{language}に厳密に従う必要があります。
        4. 冗長な出力はありません。結論の発言を含めません。
        5. トピック"{topic}"を出力しないという厳格な要件があります。
        """
        prompt = CONTENT_PROMPT.format(
            topic=topic, language=self.language, directory=self.directory)
        return await self._aask(prompt=prompt)

class TutorialAssistant(Role):
    """チュートリアルアシスタント、1文を入力してマークアップ形式のチュートリアル文書を生成します。

    Args:
        name: 役割の名前。
        profile: 役割のプロフィール説明。
        goal: 役割の目標。
        constraints: 役割の制約または要件。
        language: チュートリアル文書が生成される言語。
    """

    name: str = "Stitch"
    profile: str = "Tutorial Assistant"
    goal: str = "チュートリアル文書を生成する"
    constraints: str = "Markdownの構文に厳密に従い、整然とした標準化されたレイアウト"
    language: str = "Chinese"

    topic: str = ""
    main_title: str = ""
    total_content: str = ""

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([WriteDirectory(language=self.language)])
        self._set_react_mode(react_mode=RoleReactMode.REACT.value)

    async def _think(self) -> None:
        """役割が取るべき次のアクションを決定します。"""
        logger.info(self.rc.state)
        logger.info(self,)
        if self.rc.todo is None:
            self._set_state(0)
            return

        if self.rc.state + 1 < len(self.states):
            self._set_state(self.rc.state + 1)
        else:
            self.rc.todo = None

    async def _handle_directory(self, titles: Dict) -> Message:
        """チュートリアル文書のディレクトリを処理します。

        Args:
            titles: タイトルとディレクトリ構造を含む辞書、
                    例えば{"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}

        Returns:
            ディレクトリに関する情報を含むメッセージ。
        """
        self.main_title = titles.get("title")
        directory = f"{self.main_title}\n"
        self.total_content += f"# {self.main_title}"
        actions = list()
        for first_dir in titles.get("directory"):
            actions.append(WriteContent(
                language=self.language, directory=first_dir))
            key = list(first_dir.keys())[0]
            directory += f"- {key}\n"
            for second_dir in first_dir[key]:
                directory += f"  - {second_dir}\n"
        self.set_actions(actions)
        self.rc.todo = None
        return Message(content=directory)

    async def _act(self) -> Message:
        """役割によって決定されたアクションを実行します。

        Returns:
            アクションの結果を含むメッセージ。
        """
        todo = self.rc.todo
        if type(todo) is WriteDirectory:
            msg = self.rc.memory.get(k=1)[0]
            self.topic = msg.content
            resp = await todo.run(topic=self.topic)
            logger.info(resp)
            return await self._handle_directory(resp)
        resp = await todo.run(topic=self.topic)
        logger.info(resp)
        if self.total_content != "":
            self.total_content += "\n\n\n"
        self.total_content += resp
        return Message(content=resp, role=self.profile)

    async def _react(self) -> Message:
        """アシスタントの思考とアクションを実行します。

        Returns:
            アシスタントのアクションの最終結果を含むメッセージ。
        """
        while True:
            await self._think()
            if self.rc.todo is None:
                break
            msg = await self._act()
        root_path = TUTORIAL_PATH / datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
        await File.write(root_path, f"{self.main_title}.md", self.total_content.encode('utf-8'))
        return msg

async def main():
    msg = "AI Agent開発チュートリアル"
    role = TutorialAssistant()
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)
    return role.total_content

rtn = await main()

from bs4 import BeautifulSoup

with open("github-trending-raw.html") as f:
    html = f.read()

soup = BeautifulSoup(html, "html.parser")
for i in soup.find_all(True):
    for name in list(i.attrs):
        if i[name] and name not in ["class"]:
            del i[name]

for i in soup.find_all(["svg", "img", "video", "audio"]):
    i.decompose()

with open("github-trending-slim.html", "w") as f:
    f.write(str(soup))

import aiohttp
import asyncio
from bs4 import BeautifulSoup

def fetch_html(url):
    with open(url, encoding="utf-8") as f:
        html = f.read()
    return html

async def parse_github_trending(html):
    soup = BeautifulSoup(html, 'html.parser')

    repositories = []

    for article in soup.select('article.Box-row'):
        repo_info = {}
        
        repo_info['name'] = article.select_one('h2 a').text.strip()
        repo_info['url'] = article.select_one('h2 a')['href'].strip()

        # 説明
        description_element = article.select_one('p')
        repo_info['description'] = description_element.text.strip() if description_element else None

        # 言語
        language_element = article.select_one('span[itemprop="programmingLanguage"]')
        repo_info['language'] = language_element.text.strip() if language_element else None

        # スターとフォーク
        stars_element = article.select('a.Link--muted')[0]
        forks_element = article.select('a.Link--muted')[1]
        repo_info['stars'] = stars_element.text.strip()
        repo_info['forks'] = forks_element.text.strip()

        # 今日のスター
        today_stars_element = article.select_one('span.d-inline-block.float-sm-right')
        repo_info['today_stars'] = today_stars_element.text.strip() if today_stars_element else None

        repositories.append(repo_info)

    return repositories

async def main():
    url = 'github-trending-raw.html'
    html = fetch_html(url)
    repositories = await parse_github_trending(html)

    for repo in repositories:
        print(f"Name: {repo['name']}")
        print(f"URL: https://github.com{repo['url']}")
        print(f"Description: {repo['description']}")
        print(f"Language: {repo['language']}")
        print(f"Stars: {repo['stars']}")
        print(f"Forks: {repo['forks']}")
        print(f"Today's Stars: {repo['today_stars']}")
        print()
        
await main()

import os
from typing import Any, AsyncGenerator, Awaitable, Callable, Dict, Optional
from aiocron import crontab
from pydantic import BaseModel, Field
from pytz import BaseTzInfo

from metagpt.actions.action import Action
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message

# fix SubscriptionRunner not fully defined
from metagpt.environment import Environment as _  # noqa: F401

class SubscriptionRunner(BaseModel):
    """さまざまな役割のサブスクリプションタスクをasyncioを使用して管理するためのシンプルなラッパー。
    例:
        >>> import asyncio
        >>> from metagpt.subscription import SubscriptionRunner
        >>> from metagpt.roles import Searcher
        >>> from metagpt.schema import Message
        >>> async def trigger():
        ...     while True:
        ...         yield Message("the latest news about OpenAI")
        ...         await asyncio.sleep(3600 * 24)
        >>> async def callback(msg: Message):
        ...     print(msg.content)
        >>> async def main():
        ...     pb = SubscriptionRunner()
        ...     await pb.subscribe(Searcher(), trigger(), callback)
        ...     await pb.run()
        >>> asyncio.run(main())
    """

    tasks: Dict[Role, asyncio.Task] = Field(default_factory=dict)

    class Config:
        arbitrary_types_allowed = True

    async def subscribe(
        self,
        role: Role,
        trigger: AsyncGenerator[Message, None],
        callback: Callable[
            [
                Message,
            ],
            Awaitable[None],
        ],
    ):
        """役割をトリガーにサブスクライブし、役割の応答で呼び出されるコールバックを設定します。
        引数:
            role: サブスクライブする役割。
            trigger: 役割によって処理されるメッセージを生成する非同期ジェネレーター。
            callback: 役割からの応答で呼び出される非同期関数。
        """
        loop = asyncio.get_running_loop()

        async def _start_role():
            async for msg in trigger:
                resp = await role.run(msg)
                await callback(resp)

        self.tasks[role] = loop.create_task(_start_role(), name=f"Subscription-{role}")

    async def unsubscribe(self, role: Role):
        """役割をトリガーからサブスクライブ解除し、関連するタスクをキャンセルします。
        引数:
            role: サブスクライブ解除する役割。
        """
        task = self.tasks.pop(role)
        task.cancel()

    async def run(self, raise_exception: bool = True):
        """すべてのサブスクライブされたタスクを実行し、その完了または例外を処理します。
        引数:
            raise_exception: _description_。デフォルトはTrue。
        Raises:
            task.exception: _description_
        """
        i=0
        while True:
            i+=1
            for role, task in self.tasks.items():
                i=0
                if task.done():
                    if task.exception():
                        if raise_exception:
                            raise task.exception()
                        logger.opt(exception=task.exception()).error(
                            f"タスク {task.get_name()} 実行エラー"
                        )
                    else:
                        logger.warning(
                            f"タスク {task.get_name()} が完了しました。 "
                            "これは予期しない動作である場合は、トリガー関数を確認してください。"
                        )
                    self.tasks.pop(role)
                    break
            else:
                await asyncio.sleep(1)
            if i>0:
                break

TRENDING_ANALYSIS_PROMPT = """# 要件
あなたはGitHubトレンドアナリストであり、最新のGitHubトレンドに基づいてユーザーに洞察に満ちたパーソナライズされた推奨を提供することを目指しています。コンテキストに基づいて、次の欠落情報を埋め、魅力的で情報に富んだタイトルを生成し、ユーザーが興味に合ったリポジトリを発見できるようにします。

# 今日のGitHubトレンドに関するタイトル
## 今日のトレンド：今日のホットなGitHubプロジェクトを発見！ トレンドのプログラミング言語を探り、開発者の注目を集めている主要なドメインを発見します。 **から**まで、これまでにないトッププロジェクトを目撃してください。
## トレンドカテゴリ：今日のGitHubトレンドドメインに飛び込もう！ **や**などのドメインで特集されたプロジェクトを探り、各プロジェクトの概要を迅速に把握します。プログラミング言語、スター数などを含みます。
## リストのハイライト：新しいツール、革新的なプロジェクト、急速に人気を集めているプロジェクトを含む、GitHubトレンドの注目すべきプロジェクトを特集し、ユーザーにユニークで注目を集めるコンテンツを提供します。
---
# フォーマット例


# [タイトル]

## 今日のトレンド
今日、**と**は依然として最も人気のあるプログラミング言語として君臨しています。注目の分野には**、**、**が含まれます。
人気のプロジェクトはProject1とProject2です。

## トレンドカテゴリ
1. 生成AI
    - [Project1](https://github/xx/project1): [プロジェクトの詳細、スター数、今日の数、言語など]
    - [Project2](https://github/xx/project2): ...
...

## リストのハイライト
1. [Project1](https://github/xx/project1): [このプロジェクトが推奨される具体的な理由を提供します]。
...

---
# Githubトレンド
{trending}
"""


class CrawlOSSTrending(Action):
    async def run(self, url: str = "https://github.com/trending"):
        async with aiohttp.ClientSession() as client:
            async with client.get(url, proxy=CONFIG.global_proxy) as response:
                response.raise_for_status()
                html = await response.text()

        soup = BeautifulSoup(html, "html.parser")

        repositories = []

        for article in soup.select("article.Box-row"):
            repo_info = {}

            repo_info["name"] = (
                article.select_one("h2 a")
                .text.strip()
                .replace("\n", "")
                .replace(" ", "")
            )
            repo_info["url"] = (
                "https://github.com" + article.select_one("h2 a")["href"].strip()
            )

            # 説明
            description_element = article.select_one("p")
            repo_info["description"] = (
                description_element.text.strip() if description_element else None
            )

            # 言語
            language_element = article.select_one(
                'span[itemprop="programmingLanguage"]'
            )
            repo_info["language"] = (
                language_element.text.strip() if language_element else None
            )

            # スターとフォーク
            stars_element = article.select("a.Link--muted")[0]
            forks_element = article.select("a.Link--muted")[1]
            repo_info["stars"] = stars_element.text.strip()
            repo_info["forks"] = forks_element.text.strip()

            # 今日のスター
            today_stars_element = article.select_one(
                "span.d-inline-block.float-sm-right"
            )
            repo_info["today_stars"] = (
                today_stars_element.text.strip() if today_stars_element else None
            )

            repositories.append(repo_info)

        return repositories


class AnalysisOSSTrending(Action):
    async def run(self, trending: Any):
        return await self._aask(TRENDING_ANALYSIS_PROMPT.format(trending=trending))

# 役割実装
# V0.7以上のバージョンでは、古いバージョンのself._init_actionsをself.set_actionsに変更する必要があります
class OssWatcher(Role):
    def __init__(
        self,
        name="Codey",
        profile="OssWatcher",
        goal="洞察に満ちたGitHubトレンド分析レポートを生成する。",
        constraints="提供されたGitHubトレンドデータに基づいてのみ分析します。",
    ):
        super().__init__(name=name, profile=profile, goal=goal, constraints=constraints)
        self.set_actions([CrawlOSSTrending, AnalysisOSSTrending])
        self._set_react_mode(react_mode="by_order")

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: ready to {self.rc.todo}")
        # By choosing the Action by order under the hood
        # todo will be first SimpleWriteCode() then SimpleRunCode()
        todo = self.rc.todo

        msg = self.get_memories(k=1)[0]  # find the most k recent messages
        result = await todo.run(msg.content)

        msg = Message(content=str(result), role=self.profile, cause_by=type(todo))
        self.rc.memory.add(msg)
        return msg

async def wxpusher_callback(msg: Message):
    print(msg.content)


async def trigger():
    # ここで5回だけトリガーするように設定しましたが、while Trueで永遠に実行することもできます
    for i in range(5):
        yield Message("the latest news about OpenAI")
        await asyncio.sleep(5)
        #  5秒ごとに実行します。
        # 3600 * 24秒ごとに設定することもできます
    

# 実行エントリ
async def main():
    callbacks = []
    if not callbacks:
        async def _print(msg: Message):
            print(msg.content)
        callbacks.append(_print)

    # コールバック
    async def callback(msg):
        await asyncio.gather(*(call(msg) for call in callbacks))

    runner = SubscriptionRunner()
    await runner.subscribe(OssWatcher(), trigger(), callback)
    await runner.run()

await main()

import requests

def get_repo_info(repo_name):
    url = f"https://github.com/search?q=stars:{repo_name}&type=Repositories"
    response = requests.get(url)
    if response.status_code == 200:
        repos = []
        for repo in response.json():
            if "name" in repo and "url" in repo:
                name = repo["name"]
                url = repo["url"]
                stars = int(repo["stargazers_count"])
                forks = int(repo["forks_count"])  # Assuming the number of forks is available
                repos.append({"name": name, "url": url, "stars": stars, "forks": forks})
        return repos
    else:
        print("データの取得に失敗しました。ステータスコード:", response.status_code)

repos = get_repo_info('language')
for repo in repos:
    if 'language' not in repo or repo['language'] == None:
        del repo['language']
print(repos)

import re

import fire

from metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.team import Team


def parse_code(rsp):
    pattern = r"```python(.*)```"
    match = re.search(pattern, rsp, re.DOTALL)
    code_text = match.group(1) if match else rsp
    return code_text

class SimpleWriteCode(Action):
    PROMPT_TEMPLATE: str = """
    Write a python function that can {instruction}.
    Return ```python your_code_here ```with NO other texts,
    your code:
    """
    name: str = "SimpleWriteCode"

    async def run(self, instruction: str):
        prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)

        rsp = await self._aask(prompt)

        code_text = parse_code(rsp)

        return code_text

class SimpleWriteTest(Action):
    PROMPT_TEMPLATE: str = """
    Context: {context}
    Write {k} unit tests using pytest for the given function, assuming you have imported it.
    Return ```python your_code_here ```with NO other texts,
    your code:
    """

    name: str = "SimpleWriteTest"

    async def run(self, context: str, k: int = 3):
        prompt = self.PROMPT_TEMPLATE.format(context=context, k=k)

        rsp = await self._aask(prompt)

        code_text = parse_code(rsp)

        return code_text

class SimpleWriteReview(Action):
    PROMPT_TEMPLATE: str = """
    Context: {context}
    Review the test cases and provide one critical comments:
    """

    name: str = "SimpleWriteReview"

    async def run(self, context: str):
        prompt = self.PROMPT_TEMPLATE.format(context=context)

        rsp = await self._aask(prompt)

        return rsp

class SimpleCoder(Role):
    name: str = "Alice"
    profile: str = "SimpleCoder"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._watch([UserRequirement])
        self.set_actions([SimpleWriteCode])

class SimpleTester(Role):
    name: str = "Bob"
    profile: str = "SimpleTester"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([SimpleWriteTest])
        self._watch([SimpleWriteCode])
        # self._watch([SimpleWriteCode, SimpleWriteReview])  # feel free to try this too

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        todo = self.rc.todo

        # context = self.get_memories(k=1)[0].content # use the most recent memory as context
        context = self.get_memories()  # use all memories as context

        code_text = await todo.run(context, k=5)  # specify arguments
        msg = Message(content=code_text, role=self.profile, cause_by=type(todo))

        return msg

# 同様のプロセスでSimpleReviewerを定義します：
class SimpleReviewer(Role):
    name: str = "Charlie"
    profile: str = "SimpleReviewer"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([SimpleWriteReview])
        self._watch([SimpleWriteTest])

import fire
import typer
from metagpt.logs import logger
from metagpt.team import Team

# 非同期のメイン関数を作成
async def async_main(
    idea: str = typer.Argument(..., help="write a function that calculates the product of a list"),
    investment: float = typer.Option(default=3.0, help="Dollar amount to invest in the AI company."),
    n_round: int = typer.Option(default=5, help="Number of rounds for the simulation."),
):
    logger.info(idea)

    team = Team()
    team.hire(
        [
            SimpleCoder(),
            SimpleTester(),
            SimpleReviewer(),
        ]
    )

    team.invest(investment=investment)
    team.run_project(idea)
    await team.run(n_round=n_round)
    return team

# Jupyterで直接これを実行
await async_main(
    idea="write a function that calculates the product of a list",
    investment=3.0,
    n_round=5
)