MetaGPT

1
#!/usr/bin/env python3
2
# _*_ coding: utf-8 _*_
3
"""
4
@Time    : 2023/9/4 15:40:40
5
@Author  : Stitch-z
6
@File    : file.py
7
@Describe : General file operations.
8
"""
9
from pathlib import Path
10

11
import aiofiles
12

13
from metagpt.logs import logger
14
from metagpt.utils.exceptions import handle_exception
15

16

17
class File:
18
    """A general util for file operations."""
19

20
    CHUNK_SIZE = 64 * 1024
21

22
    @classmethod
23
    @handle_exception
24
    async def write(cls, root_path: Path, filename: str, content: bytes) -> Path:
25
        """Write the file content to the local specified path.
26

27
        Args:
28
            root_path: The root path of file, such as "/data".
29
            filename: The name of file, such as "test.txt".
30
            content: The binary content of file.
31

32
        Returns:
33
            The full filename of file, such as "/data/test.txt".
34

35
        Raises:
36
            Exception: If an unexpected error occurs during the file writing process.
37
        """
38
        root_path.mkdir(parents=True, exist_ok=True)
39
        full_path = root_path / filename
40
        async with aiofiles.open(full_path, mode="wb") as writer:
41
            await writer.write(content)
42
            logger.debug(f"Successfully write file: {full_path}")
43
            return full_path
44

45
    @classmethod
46
    @handle_exception
47
    async def read(cls, file_path: Path, chunk_size: int = None) -> bytes:
48
        """Partitioning read the file content from the local specified path.
49

50
        Args:
51
            file_path: The full file name of file, such as "/data/test.txt".
52
            chunk_size: The size of each chunk in bytes (default is 64kb).
53

54
        Returns:
55
            The binary content of file.
56

57
        Raises:
58
            Exception: If an unexpected error occurs during the file reading process.
59
        """
60
        chunk_size = chunk_size or cls.CHUNK_SIZE
61
        async with aiofiles.open(file_path, mode="rb") as reader:
62
            chunks = list()
63
            while True:
64
                chunk = await reader.read(chunk_size)
65
                if not chunk:
66
                    break
67
                chunks.append(chunk)
68
            content = b"".join(chunks)
69
            logger.debug(f"Successfully read file, the path of file: {file_path}")
70
            return content
71