use utils.read_file to replace open and read

src/linktools/_config.py

@@ -39,6 +39,7 @@
     TYPE_CHECKING, TypeVar, Type, Optional, Generator, \
     Any, Tuple, IO, Mapping, Union, List, Dict, Callable
 
+from . import utils
 from .decorator import cached_property
 from .metadata import __missing__
 from .rich import prompt, confirm, choose
@@ -158,8 +159,8 @@ def update_from_pyfile(self, filename: str, silent: bool = False) -> bool:
         d.sample = Config.Sample
         d.confirm = Config.Confirm
         try:
-            with open(filename, "rb") as config_file:
-                exec(compile(config_file.read(), filename, "exec"), d.__dict__)
+            data = utils.read_file(filename, text=False)
+            exec(compile(data, filename, "exec"), d.__dict__)
         except OSError as e:
             if silent and e.errno in (errno.ENOENT, errno.EISDIR, errno.ENOTDIR):
                 return False

src/linktools/assets/containers/100-nginx/container.py

@@ -40,9 +40,10 @@ class Container(BaseContainer):
     @cached_property
     def keys(self):
         # dnsapi.txt 内容从 https://github.com/acmesh-official/acme.sh/wiki/dnsapi 拷贝
-        with open(os.path.join(os.path.dirname(__file__), "dnsapi.txt"), "rt") as fd:
-            pattern = re.compile(r'export +(\w+)="?')
-            return sorted(list(set(pattern.findall(fd.read()))))
+        path = os.path.join(os.path.dirname(__file__), "dnsapi.txt")
+        data = utils.read_file(path, text=True)
+        pattern = re.compile(r'export +(\w+)="?')
+        return sorted(list(set(pattern.findall(data))))
 
     @cached_property
     def configs(self):

src/linktools/assets/containers/120-flare/container.py

@@ -29,7 +29,7 @@
 
 import yaml
 
-from linktools import Config
+from linktools import Config, utils
 from linktools.container import BaseContainer
 from linktools.container.container import ExposeMixin, ExposeLink, ExposeCategory
 from linktools.decorator import cached_property
@@ -58,7 +58,8 @@ def exposes(self) -> [ExposeLink]:
         return [
             self.expose_other("在线工具集合", "tools", "", "https://tool.lu/"),
             self.expose_other("在线正则表达式", "regex", "", "https://regex101.com/"),
-            self.expose_other("正则表达式手册", "regex", "", "https://tool.oschina.net/uploads/apidocs/jquery/regexp.html"),
+            self.expose_other("正则表达式手册", "regex", "",
+                              "https://tool.oschina.net/uploads/apidocs/jquery/regexp.html"),
             self.expose_other("在线json解析", "codeJson", "", "https://www.json.cn/"),
             self.expose_other("DNS查询", "dns", "", "https://tool.chinaz.com/dns/"),
             self.expose_other("图标下载", "progressDownload", "", "https://materialdesignicons.com/"),
@@ -82,34 +83,38 @@ def on_starting(self):
                         apps.append(expose)
                     bookmarks.append(expose)
 
-        with open(self.get_app_path("app", "apps.yml", create_parent=True), "wt") as fd:
-            data = {"links": []}
-            for app in apps:
+        data = {"links": []}
+        for app in apps:
+            data["links"].append({
+                "name": app.name,
+                "desc": app.desc,
+                "icon": app.icon,
+                "link": app.url,
+            })
+        utils.write_file(
+            self.get_app_path("app", "apps.yml", create_parent=True),
+            yaml.dump(data),
+        )
+
+        data = {"categories": [], "links": []}
+        for category, links in categories.items():
+            if not links:
+                continue
+            data["categories"].append({
+                "id": category.name,
+                "title": category.desc,
+            })
+            for link in links:
                 data["links"].append({
-                    "name": app.name,
-                    "desc": app.desc,
-                    "icon": app.icon,
-                    "link": app.url,
+                    "category": category.name,
+                    "name": link.name,
+                    "icon": link.icon,
+                    "link": link.url,
                 })
-            yaml.dump(data, fd)
-
-        with open(self.get_app_path("app", "bookmarks.yml", create_parent=True), "wt") as fd:
-            data = {"categories": [], "links": []}
-            for category, links in categories.items():
-                if not links:
-                    continue
-                data["categories"].append({
-                    "id": category.name,
-                    "title": category.desc,
-                })
-                for link in links:
-                    data["links"].append({
-                        "category": category.name,
-                        "name": link.name,
-                        "icon": link.icon,
-                        "link": link.url,
-                    })
-            yaml.dump(data, fd)
+        utils.write_file(
+            self.get_app_path("app", "bookmarks.yml", create_parent=True),
+            yaml.dump(data),
+        )
 
         self.manager.change_owner(
             self.get_app_path("app"),
@@ -119,5 +124,4 @@ def on_starting(self):
         self.write_nginx_conf(
             self.manager.config.get("FLARE_DOAMIN"),
             self.get_path("nginx.conf"),
-            name="flare",
         )

src/linktools/cli/commands/common/cert.py

@@ -87,7 +87,7 @@ def subject_name_hash_old(cert: OpenSSL.SSL.X509):
 
         cert = OpenSSL.crypto.load_certificate(
             OpenSSL.crypto.FILETYPE_PEM,
-            utils.read_file(os.path.expanduser(args.path), binary=True)
+            utils.read_file(os.path.expanduser(args.path), text=False)
         )
 
         issuer = cert.get_issuer()

src/linktools/cli/commands/common/grep.py

@@ -111,7 +111,7 @@ def match(self, path: str):
     def on_file(self, filename: str):
         if os.path.exists(filename):
             try:
-                with open(filename, 'rb') as fd:
+                with open(filename, "rb") as fd:
                     buffer = fd.read(1024)
                 mimetype = magic.from_buffer(buffer, mime=True)
                 if not GrepHandler.handle(self, filename, mimetype):

src/linktools/cli/device.py

@@ -34,6 +34,7 @@
 from typing import Optional, Callable, List, Type, Generic
 
 from . import BaseCommand
+from .. import utils
 from ..android import Adb, AdbError, Device as AdbDevice
 from ..device import Bridge, BridgeError, BaseDevice, BridgeType, DeviceType
 from ..ios import Sib, SibError, Device as SibDevice
@@ -48,13 +49,11 @@ def __init__(self, path: str):
 
     def read(self) -> Optional[str]:
         if os.path.exists(self.path):
-            with open(self.path, "rt") as fd:
-                return fd.read().strip()
+            return utils.read_file(self.path, text=True).strip()
         return None
 
     def write(self, cache: str) -> None:
-        with open(self.path, "wt") as fd:
-            fd.write(cache)
+        utils.write_file(self.path, cache)
 
     def __call__(self, fn: Callable[..., BaseDevice]):
         @functools.wraps(fn)

src/linktools/container/container.py

@@ -236,8 +236,10 @@ def get_docker_compose_file(self) -> Optional[str]:
                 f"{self.name}.yml",
                 create_parent=True,
             )
-            with open(destination, "wt") as fd:
-                yaml.dump(self.docker_compose, fd)
+            utils.write_file(
+                destination,
+                yaml.dump(self.docker_compose)
+            )
         return destination
 
     def get_docker_file_path(self) -> Optional[str]:
@@ -249,8 +251,10 @@ def get_docker_file_path(self) -> Optional[str]:
                 f"{self.name}.Dockerfile",
                 create_parent=True,
             )
-            with open(destination, "wt") as fd:
-                fd.write(self.docker_file)
+            utils.write_file(
+                destination,
+                self.docker_file
+            )
         return destination
 
     @subcommand("shell", help="exec into container using command sh", prefix_chars=chr(0))
@@ -403,12 +407,10 @@ def render_template(self, source: str, destination: str = None, **kwargs: Any):
         context.setdefault("int", lambda obj, default=0: self.manager.config.cast(obj, type=int, default=default))
         context.setdefault("float", lambda obj, default=0.0: self.manager.config.cast(obj, type=float, default=default))
 
-        with open(source, "rt") as fd:
-            template = Template(fd.read())
+        template = Template(utils.read_file(source, text=True))
         result = template.render(context)
-        if destination is not None:
-            with open(destination, "wt") as fd:
-                fd.write(result)
+        if destination:
+            utils.write_file(destination, result)
 
         return result
 

src/linktools/container/manager.py

@@ -529,15 +529,18 @@ def _repo_config_path(self):
     def _load_config(self, path: str) -> Union[Dict, List, Tuple]:
         if os.path.exists(path):
             try:
-                with open(path, "rt") as fd:
-                    return json.load(fd)
+                return json.loads(
+                    utils.read_file(path, text=True)
+                )
             except Exception as e:
                 self.logger.warning(f"Failed to load config file {path}: {e}")
         return {}
 
     def _dump_config(self, path: str, config: Union[Dict, List, Tuple]):
         try:
-            with open(path, "wt") as fd:
-                json.dump(config, fd, indent=2, ensure_ascii=False)
+            utils.write_file(
+                path,
+                json.dumps(config, indent=2, ensure_ascii=False)
+            )
         except Exception as e:
             self.logger.warning(f"Failed to dump config file {path}: {e}")

src/linktools/frida/script.py

@@ -96,9 +96,8 @@ def filename(self):
         return self._path
 
     def _load(self) -> Optional[str]:
-        with open(self._path, "rb") as f:
-            _logger.info(f"Load {self}")
-            return f.read().decode("utf-8")
+        _logger.info(f"Load {self}")
+        return utils.read_file(self._path, text=True)
 
 
 class FridaEvalCode(FridaUserScript):
@@ -137,18 +136,15 @@ def _load(self):
             _logger.info(f"Download {self}")
             target_path = file.save()
 
-            with open(target_path, "rb") as f:
-                source = f.read().decode("utf-8")
-
+            source = utils.read_file(target_path, text=True)
             if self._trusted:
                 _logger.info(f"Load trusted {self}")
                 return source
 
             cached_md5 = ""
             cached_md5_path = target_path + ".md5"
             if os.path.exists(cached_md5_path):
-                with open(cached_md5_path, "rt") as fd:
-                    cached_md5 = fd.read()
+                cached_md5 = utils.read_file(cached_md5_path, text=True)
 
             source_md5 = utils.get_md5(source)
             if cached_md5 == source_md5:
@@ -169,8 +165,7 @@ def _load(self):
                      f"Source: {os.linesep}{source_summary}{os.linesep}" \
                      f"Are you sure you'd like to trust it?"
             if confirm(prompt):
-                with open(cached_md5_path, "wt") as fd:
-                    fd.write(source_md5)
+                utils.write_file(cached_md5_path, source_md5)
                 _logger.info(f"Load trusted {self}")
                 return source
             else: