🚧 build

pdm.lock

@@ -6,7 +6,7 @@ groups = ["default", "sqlite"]
 cross_platform = true
 static_urls = false
 lock_version = "4.3"
-content_hash = "sha256:d6e8399bd56ac79d004edf42f2d700c865dfc865dbb7a9d6bfab690bcc18e2e4"
+content_hash = "sha256:0f64786cb90cdd13b732f315e7a378ca8529bee23c238e04daaf69518f74211b"
 
 [[package]]
 name = "aiodns"
@@ -93,19 +93,6 @@ files = [
     {file = "aiohttp-3.8.5.tar.gz", hash = "sha256:b9552ec52cc147dbf1944ac7ac98af7602e51ea2dcd076ed194ca3c0d1c7d0bc"},
 ]
 
-[[package]]
-name = "aiomysql"
-version = "0.2.0"
-requires_python = ">=3.7"
-summary = "MySQL driver for asyncio."
-dependencies = [
-    "PyMySQL>=1.0",
-]
-files = [
-    {file = "aiomysql-0.2.0-py3-none-any.whl", hash = "sha256:b7c26da0daf23a5ec5e0b133c03d20657276e4eae9b73e040b72787f6f6ade0a"},
-    {file = "aiomysql-0.2.0.tar.gz", hash = "sha256:558b9c26d580d08b8c5fd1be23c5231ce3aeff2dadad989540fee740253deb67"},
-]
-
 [[package]]
 name = "aiosignal"
 version = "1.3.1"
@@ -446,18 +433,6 @@ files = [
     {file = "peewee-3.16.3.tar.gz", hash = "sha256:12b30e931193bc37b11f7c2ac646e3f67125a8b1a543ad6ab37ad124c8df7d16"},
 ]
 
-[[package]]
-name = "peewee-async"
-version = "0.8.1"
-summary = "Asynchronous interface for peewee ORM powered by asyncio."
-dependencies = [
-    "peewee<4.0,>=3.15.4",
-]
-files = [
-    {file = "peewee-async-0.8.1.tar.gz", hash = "sha256:2ecc99fbee5b4be6e8423dd851996249adda2111ee5e7b60ba666ede2f796cd0"},
-    {file = "peewee_async-0.8.1-py3-none-any.whl", hash = "sha256:e08a09fa9f104c352342abaf7f5bff6327490f5b931552f27bc1aac26b58ad44"},
-]
-
 [[package]]
 name = "pycares"
 version = "4.3.0"
@@ -587,16 +562,6 @@ files = [
     {file = "pydantic_settings-2.0.3.tar.gz", hash = "sha256:962dc3672495aad6ae96a4390fac7e593591e144625e5112d359f8f67fb75945"},
 ]
 
-[[package]]
-name = "pymysql"
-version = "1.1.0"
-requires_python = ">=3.7"
-summary = "Pure Python MySQL Driver"
-files = [
-    {file = "PyMySQL-1.1.0-py3-none-any.whl", hash = "sha256:8969ec6d763c856f7073c4c64662882675702efcb114b4bcbb955aea3a069fa7"},
-    {file = "PyMySQL-1.1.0.tar.gz", hash = "sha256:4f13a7df8bf36a51e81dd9f3605fede45a4878fe02f9236349fd82a3f0612f96"},
-]
-
 [[package]]
 name = "python-dotenv"
 version = "1.0.0"

pyproject.toml

@@ -13,9 +13,7 @@ dependencies = [
     "regex<2023.0.0,>=2022.10.31",
     "arko-wrapper<1.0.0,>=0.2.8",
     "peewee>=3.16.3",
-    "peewee-async>=0.8.1",
     "pydantic>=2.3.0",
-    "aiomysql>=0.2.0",
     "pydantic-settings>=2.0.3",
     "Cython>=3.0.1",
     "apsw>=3.43.0.0",

src/genshin/wiki/_database/_mode.py

@@ -0,0 +1,83 @@
+from pathlib import Path
+
+import peewee
+from peewee import IntegerField, SqliteDatabase
+
+from genshin.wiki.config import get_wiki_lang
+from genshin.wiki.tools.const import DATA_DIR
+from genshin.wiki.tools.typedefs import Lang
+from genshin.wiki.utils import LimitedSizeDict
+
+__all__ = (
+    "Model",
+    "ModelMeta",
+    "MapString",
+    "MapStringField",
+)
+
+
+_database = SqliteDatabase(Path(__file__).joinpath('../sqlite.db'))
+_database.connect()
+
+_lang_database_map: LimitedSizeDict[Lang, SqliteDatabase] = LimitedSizeDict(size_limit=256)
+
+class ModelMeta:
+    database: SqliteDatabase = _database
+
+
+class Model(peewee.Model):
+    class Meta(ModelMeta):
+        abstract = True
+
+_map_string_cache: dict[int, "MapString"] = {}
+
+class MapString(str):
+    __slots__ = ("_text_id", "_lang")
+
+    def __new__(cls, target: int | str) -> "MapString":
+        lang = get_wiki_lang()
+
+        _map_string_cache_key = hash((str(lang), target,))
+        result = _map_string_cache.get(_map_string_cache_key, None)
+        if result is not None:
+            return result
+
+        if lang not in _lang_database_map:
+            database = SqliteDatabase(DATA_DIR.joinpath(lang + '.db').resolve())
+            database.connect()
+            _lang_database_map[lang] = database
+        else:
+            database = _lang_database_map[lang]
+
+        if isinstance(target, int):
+            text_id = target
+            text = database.execute_sql(
+                "SELECT context FROM mapping_text WHERE id = ?", (target,)
+            ).fetchone()[0]
+        else:
+            text = target
+            text_id = database.execute_sql(
+                "SELECT id FROM mapping_text WHERE context = ?", (target,)
+            ).fetchone()[0]
+        
+        result = super().__new__(cls, text)
+        result._text_id = text_id
+        result._lang = lang
+        _map_string_cache[_map_string_cache_key] = result
+        return result
+    
+    @property
+    def lang(self) -> Lang:
+        return self._lang
+
+    @property
+    def text_id(self) -> int:
+        return self._text_id
+
+class MapStringField(IntegerField):
+
+    def db_value(self, value: str | int | MapString) -> int:
+        return MapString(value).text_id
+
+    def python_value(self, value: int) -> MapString:
+        return MapString(value)

src/genshin/wiki/_database/item.py

@@ -1,6 +1,10 @@
 from peewee import IntegerField, CharField
-from genshin.wiki._mode import Model
+from genshin.wiki._database._mode import Model, MapStringField
 
 
 class Item(Model):
-    name = CharField()
+    name = MapStringField()
+    """Name of the item."""
+
+    description = MapStringField()
+    """Description of the item."""

src/genshin/wiki/_mode.py

@@ -1,38 +0,0 @@
-from typing import TYPE_CHECKING, TypeVar
-
-import peewee
-from peewee import CharField, SqliteDatabase
-from peewee_async import MySQLDatabase
-
-from genshin.wiki.config import genshin_wiki_config
-
-if TYPE_CHECKING:
-    from peewee import Database
-
-__all__ = (
-    "Model",
-    "ModelMeta",
-)
-
-
-T = TypeVar("T")
-
-
-database = MySQLDatabase(genshin_wiki_config.database_url)
-database.connect()
-
-class ModelMeta:
-    database: "Database" = database
-
-
-class Model(peewee.Model):
-    class Meta(ModelMeta):
-        abstract = True
-
-class MapStringField(CharField):
-
-    def db_value(self, value):
-        return value.hex  # convert UUID to hex string.
-
-    def python_value(self, value):
-        return value

src/genshin/wiki/config.py

@@ -1,55 +1,72 @@
-import inspect
 from contextlib import contextmanager
-from typing import Any, Generator
+from contextvars import ContextVar
+from typing import Any, Generator, Optional, TYPE_CHECKING
 
+from pydantic import PrivateAttr
 from pydantic_settings import BaseSettings
 
 from genshin.wiki.tools.typedefs import Lang
 
-__all__ = ("genshin_wiki_config", "use_genshin_wiki_config")
+if TYPE_CHECKING:
+    from contextvars import Token
+
+__all__ = ("set_wiki_lang", "get_wiki_lang",)
 
 
 class _GenshinWikiConfig(BaseSettings, env_prefix='genshin_wiki_'):
+    _token: Optional["Token"] = PrivateAttr(None)
+
     lang: Lang = "chs"
     """Language of the data to be used."""
 
-    database_url: str = "mysql://root:123456@localhost:3306/genshin_wiki?charset=utf8mb4"
-    """The connection url of the database."""
-
     metadata_repo: str = "https://gitlab.com/Dimbreath/AnimeGameData/"
     """The repo link of the GenshinData."""
 
-genshin_wiki_config = _GenshinWikiConfig()
+_genshin_wiki_config = _GenshinWikiConfig()
-
+_genshin_wiki_lang: ContextVar[Lang] = ContextVar('_genshin_wiki_lang', default=_genshin_wiki_config.lang)
 
 @contextmanager
-def use_genshin_wiki_config(**kwargs) -> Generator[_GenshinWikiConfig, Any, None]:
+def set_wiki_lang(lang: Lang) -> Generator[_GenshinWikiConfig, Any, None]:
-    global genshin_wiki_config
+    token = _genshin_wiki_lang.set(lang)
-    old_config = genshin_wiki_config
+    config = _GenshinWikiConfig(lang=lang, metadata_repo=_genshin_wiki_config.metadata_repo)
-    config_dict = genshin_wiki_config.model_dump()
+    try:
-    config_dict.update(kwargs)
+        yield config
-    new_config = _GenshinWikiConfig(**config_dict)
+    finally:
+        _genshin_wiki_lang.reset(token)
 
-    frame = inspect.currentframe().f_back.f_back
+def get_wiki_lang() -> Lang:
+    return _genshin_wiki_lang.get()
 
-    local_keys = []
+# @contextmanager
-    for k, v in frame.f_locals.items():
+# def use_genshin_wiki_config(**kwargs) -> "Generator[_GenshinWikiConfig, Any, None]":
-        if type(v).__name__ == _GenshinWikiConfig.__name__:
+#     import inspect
-            frame.f_locals[k] = new_config
+#     from contextlib import contextmanager
-            local_keys.append(k)
+#     global genshin_wiki_config
-    global_keys = []
+#     old_config = genshin_wiki_config
-    for k, v in frame.f_globals.items():
+#     config_dict = genshin_wiki_config.model_dump()
-        if type(v).__name__ == _GenshinWikiConfig.__name__:
+#     config_dict.update(kwargs)
-            frame.f_globals[k] = new_config
+#     new_config = _GenshinWikiConfig(**config_dict)
-            global_keys.append(k)
+#
-
+#     frame = inspect.currentframe().f_back.f_back
-    genshin_wiki_config = _GenshinWikiConfig(**config_dict)
+#
-
+#     local_keys = []
-    yield genshin_wiki_config
+#     for k, v in frame.f_locals.items():
-
+#         if type(v).__name__ == _GenshinWikiConfig.__name__:
-    genshin_wiki_config = old_config
+#             frame.f_locals[k] = new_config
-
+#             local_keys.append(k)
-    for k in local_keys:
+#     global_keys = []
-        frame.f_locals[k] = old_config
+#     for k, v in frame.f_globals.items():
-    for k in global_keys:
+#         if type(v).__name__ == _GenshinWikiConfig.__name__:
-        frame.f_globals[k] = old_config
+#             frame.f_globals[k] = new_config
+#             global_keys.append(k)
+#
+#     genshin_wiki_config = _GenshinWikiConfig(**config_dict)
+#
+#     yield genshin_wiki_config
+#
+#     genshin_wiki_config = old_config
+#
+#     for k in local_keys:
+#         frame.f_locals[k] = old_config
+#     for k in global_keys:
+#         frame.f_globals[k] = old_config

src/genshin/wiki/manager.py

@@ -5,7 +5,7 @@ import orjson as json
 from aiofiles import open as async_open
 from pydantic import Json
 
-from genshin.wiki.config import genshin_wiki_config
+from genshin.wiki.config import get_wiki_lang
 from genshin.wiki.tools.const import DATA_DIR
 from genshin.wiki.tools.typedefs import Lang
 from genshin.wiki.utils.funcs import get_repo_raw
@@ -13,60 +13,3 @@ from genshin.wiki.utils.net import Net
 
 LANG_DATA_DIR = DATA_DIR.joinpath("lang")
 database_map: dict[Lang, SqliteDatabase] = {}
-
-
-class MappingText:
-    _id: int | None = None
-    _context: str | None = None
-
-    def __init__(self, target: int | str) -> None:
-        if genshin_wiki_config.lang not in database_map:
-            self._database = SqliteDatabase(
-                f"sqlite:///{LANG_DATA_DIR.joinpath(genshin_wiki_config.lang)}.sqlite"
-            )
-            database_map[genshin_wiki_config.lang] = self._database
-        else:
-            self._database = database_map[genshin_wiki_config.lang]
-
-        if isinstance(target, int):
-            self._id = target
-            self._context = self._database.execute_sql(
-                "SELECT context FROM mapping_text WHERE id = ?", (target,)
-            ).fetchone()[0]
-        else:
-            self._context = target
-            self._id = self._database.execute_sql(
-                "SELECT id FROM mapping_text WHERE context = ?", (target,)
-            ).fetchone()[0]
-
-    def __eq__(self, other) -> bool:
-        if isinstance(other, MappingText):
-            return self._id == other._id
-        return False
-
-    def __hash__(self) -> int:
-        return self._id
-
-
-class ResourceManager(Net):
-    @property
-    def lang(self) -> Lang:
-        return genshin_wiki_config.lang
-
-    async def metadata(
-        self, file_path: str, *, overwritten: bool = False
-    ) -> Json[dict[str, Any]]:
-        """Download metadata from the GenshinData repo."""
-        url = get_repo_raw(genshin_wiki_config.metadata_repo) + file_path
-        if (path := DATA_DIR.joinpath(file_path)).exists():
-            if not overwritten:
-                async with async_open(path, "r") as f:
-                    return json.loads(await f.read())
-            else:
-                async with async_open(path, "w") as f:
-                    async with await self._get(url) as resp:
-                        await f.write(await resp.text())
-                    return await resp.json()
-
-    async def text(self, text_id: int) -> str:
-        ...

src/genshin/wiki/utils/__init__.py

@@ -0,0 +1,23 @@
+from collections import OrderedDict
+from typing import TypeVar
+
+__all__ = ("LimitedSizeDict", )
+
+K = TypeVar('K')
+V = TypeVar('V')
+
+
+class LimitedSizeDict(OrderedDict[K, V]):
+    def __init__(self, *args, size_limit: int | None = None, **kwargs) -> None:
+        self.size_limit = size_limit
+        OrderedDict.__init__(self, *args, **kwargs)
+        self._check_size_limit()
+
+    def __setitem__(self, key: K, value: V) -> None:
+        OrderedDict.__setitem__(self, key, value)
+        self._check_size_limit()
+
+    def _check_size_limit(self) -> None:
+        if self.size_limit is not None:
+            while len(self) > self.size_limit:
+                self.popitem(last=False)