Automatically fix permissions issues

Author: monosans

poetry.lock

@@ -4,7 +4,7 @@
 
 # Monkeypatch os.link to make aiofiles work on Termux
 if not hasattr(_os, "link"):
-    from .typing_compat import Any as _Any
+    from typing_extensions import Any as _Any
 
     def _link(*args: _Any, **kwargs: _Any) -> None:  # noqa: ARG001
         raise RuntimeError

proxy_scraper_checker/__init__.py

@@ -13,11 +13,11 @@
 from rich.logging import RichHandler
 from rich.progress import BarColumn, MofNCompleteColumn, Progress, TextColumn
 from rich.table import Table
+from typing_extensions import Any
 
 from . import checker, geodb, http, output, scraper, sort, utils
 from .settings import Settings
 from .storage import ProxyStorage
-from .typing_compat import Any
 
 if sys.version_info >= (3, 11):
     try:

proxy_scraper_checker/__main__.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+import platformdirs
+
+logger = logging.getLogger(__name__)
+CACHE_PATH = platformdirs.user_cache_path("proxy_scraper_checker")
+
+
+def add_permission(path: Path, permission: int, /) -> None:
+    current_permissions = path.stat().st_mode
+    new_permissions = current_permissions | permission
+    if current_permissions != new_permissions:
+        path.chmod(new_permissions)
+        logger.info(
+            "Changed permissions of %s from %o to %o",
+            path,
+            current_permissions,
+            new_permissions,
+        )
+
+
+def maybe_add_permission(path: Path, permission: int, /) -> None:
+    try:
+        add_permission(path, permission)
+    except FileNotFoundError:
+        pass
+
+
+def create_or_fix_dir(path: Path, /, *, permissions: int) -> None:
+    try:
+        path.mkdir(parents=True)
+    except FileExistsError:
+        if not path.is_dir():
+            msg = f"{path} is not a directory"
+            raise ValueError(msg) from None
+        add_permission(path, permissions)

proxy_scraper_checker/cache.py

@@ -1,26 +1,26 @@
 from __future__ import annotations
 
 import logging
-from pathlib import Path
+import stat
 from typing import Optional
 
 import aiofiles
-import aiofiles.ospath
 from aiohttp import ClientResponse, ClientSession, hdrs
 from rich.progress import Progress, TaskID
 
-from . import cache
-from .utils import IS_DOCKER, bytes_decode
+from . import fs
+from .utils import IS_DOCKER, asyncify, bytes_decode
 
 logger = logging.getLogger(__name__)
 
 GEODB_URL = "https://raw.githubusercontent.com/P3TERX/GeoLite.mmdb/download/GeoLite2-City.mmdb"
-GEODB_PATH = Path(cache.DIR, "geolocation_database.mmdb")
+GEODB_PATH = fs.CACHE_PATH / "geolocation_database.mmdb"
 GEODB_ETAG_PATH = GEODB_PATH.with_suffix(".mmdb.etag")
 
 
 async def _read_etag() -> Optional[str]:
     try:
+        await asyncify(fs.add_permission)(GEODB_ETAG_PATH, stat.S_IRUSR)
         async with aiofiles.open(GEODB_ETAG_PATH, "rb") as etag_file:
             content = await etag_file.read()
     except FileNotFoundError:
@@ -29,6 +29,7 @@ async def _read_etag() -> Optional[str]:
 
 
 async def _save_etag(etag: str, /) -> None:
+    await asyncify(fs.maybe_add_permission)(GEODB_ETAG_PATH, stat.S_IWUSR)
     async with aiofiles.open(
         GEODB_ETAG_PATH, "w", encoding="utf-8"
     ) as etag_file:
@@ -38,6 +39,7 @@ async def _save_etag(etag: str, /) -> None:
 async def _save_geodb(
     *, progress: Progress, response: ClientResponse, task: TaskID
 ) -> None:
+    await asyncify(fs.maybe_add_permission)(GEODB_PATH, stat.S_IWUSR)
     async with aiofiles.open(GEODB_PATH, "wb") as geodb:
         async for chunk in response.content.iter_any():
             await geodb.write(chunk)
@@ -47,7 +49,7 @@ async def _save_geodb(
 async def download_geodb(*, progress: Progress, session: ClientSession) -> None:
     headers = (
         {hdrs.IF_NONE_MATCH: current_etag}
-        if await aiofiles.ospath.exists(GEODB_PATH)
+        if await asyncify(GEODB_PATH.exists)()
         and (current_etag := await _read_etag())
         else None
     )

proxy_scraper_checker/fs.py

@@ -2,19 +2,19 @@
 
 import json
 import logging
+import stat
 from shutil import rmtree
 from typing import Sequence, Union
 
-import aiofiles.ospath
 import maxminddb
 
-from . import sort
+from . import fs, sort
 from .geodb import GEODB_PATH
 from .null_context import NullContext
 from .proxy import Proxy
 from .settings import Settings
 from .storage import ProxyStorage
-from .utils import IS_DOCKER
+from .utils import IS_DOCKER, asyncify
 
 logger = logging.getLogger(__name__)
 
@@ -30,14 +30,16 @@ def _create_proxy_list_str(
     )
 
 
-@aiofiles.ospath.wrap
+@asyncify
 def save_proxies(*, settings: Settings, storage: ProxyStorage) -> None:
     if settings.output_json:
-        mmdb: Union[maxminddb.Reader, NullContext] = (
-            maxminddb.open_database(GEODB_PATH)
-            if settings.enable_geolocation
-            else NullContext()
-        )
+        if settings.enable_geolocation:
+            fs.add_permission(GEODB_PATH, stat.S_IRUSR)
+            mmdb: Union[maxminddb.Reader, NullContext] = (
+                maxminddb.open_database(GEODB_PATH)
+            )
+        else:
+            mmdb = NullContext()
         with mmdb as mmdb_reader:
             proxy_dicts = [
                 {
@@ -54,16 +56,19 @@ def save_proxies(*, settings: Settings, storage: ProxyStorage) -> None:
                 }
                 for proxy in sorted(storage, key=sort.timeout_sort_key)
             ]
-            with (settings.output_path / "proxies.json").open(
-                "w", encoding="utf-8"
-            ) as f:
-                json.dump(
-                    proxy_dicts, f, ensure_ascii=False, separators=(",", ":")
-                )
-            with (settings.output_path / "proxies_pretty.json").open(
-                "w", encoding="utf-8"
-            ) as f:
-                json.dump(proxy_dicts, f, ensure_ascii=False, indent="\t")
+            for path, indent, separators in (
+                (settings.output_path / "proxies.json", None, (",", ":")),
+                (settings.output_path / "proxies_pretty.json", "\t", None),
+            ):
+                path.unlink(missing_ok=True)
+                with path.open("w", encoding="utf-8") as f:
+                    json.dump(
+                        proxy_dicts,
+                        f,
+                        ensure_ascii=False,
+                        indent=indent,
+                        separators=separators,
+                    )
     if settings.output_txt:
         sorted_proxies = sorted(storage, key=settings.sorting_key)
         grouped_proxies = tuple(
@@ -78,7 +83,7 @@ def save_proxies(*, settings: Settings, storage: ProxyStorage) -> None:
                 rmtree(folder)
             except FileNotFoundError:
                 pass
-            folder.mkdir(parents=True, exist_ok=True)
+            folder.mkdir()
             text = _create_proxy_list_str(
                 proxies=sorted_proxies,
                 anonymous_only=anonymous_only,

proxy_scraper_checker/geodb.py

@@ -5,7 +5,7 @@
 import json
 import logging
 import math
-import os
+import stat
 import sys
 from pathlib import Path
 from typing import (
@@ -25,13 +25,13 @@
 import platformdirs
 from aiohttp import ClientSession, ClientTimeout
 from aiohttp_socks import ProxyType
+from typing_extensions import Any, Literal, Self
 
-from . import cache, sort
+from . import fs, sort
 from .http import get_response_text
 from .null_context import NullContext
 from .parsers import parse_ipv4
-from .typing_compat import Any, Literal, Self
-from .utils import IS_DOCKER, create_or_check_dir
+from .utils import IS_DOCKER, asyncify
 
 if TYPE_CHECKING:
     from .proxy import Proxy
@@ -269,12 +269,17 @@ async def from_mapping(
         output_path = (
             platformdirs.user_data_path("proxy_scraper_checker")
             if IS_DOCKER
-            else cfg["output"]["path"]
+            else Path(cfg["output"]["path"])
         )
 
         _, _, (check_website_type, real_ip) = await asyncio.gather(
-            create_or_check_dir(output_path, mode=os.W_OK | os.X_OK),
-            create_or_check_dir(cache.DIR, mode=os.R_OK | os.W_OK | os.X_OK),
+            asyncify(fs.create_or_fix_dir)(
+                output_path, permissions=stat.S_IXUSR | stat.S_IWUSR
+            ),
+            asyncify(fs.create_or_fix_dir)(
+                fs.CACHE_PATH,
+                permissions=stat.S_IRUSR | stat.S_IXUSR | stat.S_IWUSR,
+            ),
             _get_check_website_type_and_real_ip(
                 check_website=cfg["check_website"], session=session
             ),

proxy_scraper_checker/output.py

@@ -1,14 +1,16 @@
 from __future__ import annotations
 
 import asyncio
+import functools
 import os
-from pathlib import Path
-from typing import Union
+from typing import Callable
 from urllib.parse import urlparse
 
-import aiofiles.os
-import aiofiles.ospath
 import charset_normalizer
+from typing_extensions import ParamSpec, TypeVar
+
+T = TypeVar("T")
+P = ParamSpec("P")
 
 IS_DOCKER = os.getenv("IS_DOCKER") == "1"
 
@@ -22,14 +24,10 @@ def bytes_decode(value: bytes, /) -> str:
     return str(charset_normalizer.from_bytes(value)[0])
 
 
-async def create_or_check_dir(path: Union[Path, str], /, *, mode: int) -> None:
-    try:
-        await aiofiles.os.makedirs(path)
-    except FileExistsError:
-        access_task = asyncio.create_task(aiofiles.os.access(path, mode))
-        if not await aiofiles.ospath.isdir(path):
-            msg = f"{path} is not a directory"
-            raise ValueError(msg) from None
-        if not await access_task:
-            msg = f"{path} is not accessible"
-            raise ValueError(msg) from None
+def asyncify(f: Callable[P, T], /) -> Callable[P, asyncio.Future[T]]:
+    def wrapper(*args: P.args, **kwargs: P.kwargs) -> asyncio.Future[T]:
+        return asyncio.get_running_loop().run_in_executor(
+            None, functools.partial(f, *args, **kwargs)
+        )
+
+    return functools.update_wrapper(wrapper, f)

proxy_scraper_checker/settings.py

@@ -26,7 +26,7 @@ maxminddb = ">=1.3,<3"
 platformdirs = "<5"
 rich = ">=12.3,<14"
 tomli = { version = "<3", python = "<3.11" }
-typing-extensions = { version = "^4.4", python = "<3.11" }
+typing-extensions = "^4.4"
 uvloop = { version = ">=0.14,<0.20", optional = true, markers = "implementation_name == 'cpython' and (sys_platform == 'darwin' or sys_platform == 'linux')" }
 
 [tool.poetry.extras]

proxy_scraper_checker/typing_compat.py

@@ -51,7 +51,6 @@ ignore = [
 ]
 ignore-init-module-imports = true
 select = ["ALL"]
-typing-modules = ["proxy_scraper_checker.typing_compat"]
 
 [lint.flake8-self]
 ignore-names = []