mirror of
https://github.com/ankitects/anki.git
synced 2025-09-19 06:22:22 -04:00
131 lines
4.4 KiB
Python
131 lines
4.4 KiB
Python
# Copyright: Ankitects Pty Ltd and contributors
|
|
# License: GNU AGPL, version 3 or later; http://www.gnu.org/licenses/agpl.html
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, List, Optional, Sequence, Union
|
|
from weakref import ref
|
|
|
|
import anki.buildinfo
|
|
from anki._backend.generated import RustBackendGenerated
|
|
from anki.dbproxy import Row as DBRow
|
|
from anki.dbproxy import ValueForDB
|
|
from anki.errors import backend_exception_to_pylib
|
|
from anki.utils import from_json_bytes, to_json_bytes
|
|
|
|
from . import backend_pb2 as pb
|
|
from . import rsbridge
|
|
from .fluent import GeneratedTranslations, LegacyTranslationEnum
|
|
|
|
# pylint: disable=c-extension-no-member
|
|
assert rsbridge.buildhash() == anki.buildinfo.buildhash
|
|
|
|
|
|
class RustBackend(RustBackendGenerated):
|
|
"""
|
|
Python bindings for Anki's Rust libraries.
|
|
|
|
Please do not access methods on the backend directly - they may be changed
|
|
or removed at any time. Instead, please use the methods on the collection
|
|
instead. Eg, don't use col._backend.all_deck_config(), instead use
|
|
col.decks.all_config()
|
|
|
|
If you need to access a backend method that is not currently accessible
|
|
via the collection, please send through a pull request that adds a
|
|
public method.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
langs: Optional[List[str]] = None,
|
|
server: bool = False,
|
|
) -> None:
|
|
# pick up global defaults if not provided
|
|
if langs is None:
|
|
langs = [anki.lang.currentLang]
|
|
|
|
init_msg = pb.BackendInit(
|
|
preferred_langs=langs,
|
|
server=server,
|
|
)
|
|
self._backend = rsbridge.open_backend(init_msg.SerializeToString())
|
|
|
|
def db_query(
|
|
self, sql: str, args: Sequence[ValueForDB], first_row_only: bool
|
|
) -> List[DBRow]:
|
|
return self._db_command(
|
|
dict(kind="query", sql=sql, args=args, first_row_only=first_row_only)
|
|
)
|
|
|
|
def db_execute_many(self, sql: str, args: List[List[ValueForDB]]) -> List[DBRow]:
|
|
return self._db_command(dict(kind="executemany", sql=sql, args=args))
|
|
|
|
def db_begin(self) -> None:
|
|
return self._db_command(dict(kind="begin"))
|
|
|
|
def db_commit(self) -> None:
|
|
return self._db_command(dict(kind="commit"))
|
|
|
|
def db_rollback(self) -> None:
|
|
return self._db_command(dict(kind="rollback"))
|
|
|
|
def _db_command(self, input: Dict[str, Any]) -> Any:
|
|
try:
|
|
return from_json_bytes(self._backend.db_command(to_json_bytes(input)))
|
|
except Exception as e:
|
|
err_bytes = bytes(e.args[0])
|
|
err = pb.BackendError()
|
|
err.ParseFromString(err_bytes)
|
|
raise backend_exception_to_pylib(err)
|
|
|
|
def translate(
|
|
self, key: Union[LegacyTranslationEnum, int], **kwargs: Union[str, int, float]
|
|
) -> str:
|
|
int_key = key if isinstance(key, int) else key.value
|
|
return self.translate_string(translate_string_in(key=int_key, **kwargs))
|
|
|
|
def format_time_span(
|
|
self,
|
|
seconds: Any,
|
|
context: Any = 2,
|
|
) -> str:
|
|
print(
|
|
"please use col.format_timespan() instead of col.backend.format_time_span()"
|
|
)
|
|
return self.format_timespan(seconds=seconds, context=context)
|
|
|
|
def _run_command(self, service: int, method: int, input: Any) -> bytes:
|
|
input_bytes = input.SerializeToString()
|
|
try:
|
|
return self._backend.command(service, method, input_bytes)
|
|
except Exception as e:
|
|
err_bytes = bytes(e.args[0])
|
|
err = pb.BackendError()
|
|
err.ParseFromString(err_bytes)
|
|
raise backend_exception_to_pylib(err)
|
|
|
|
|
|
def translate_string_in(
|
|
key: int, **kwargs: Union[str, int, float]
|
|
) -> pb.TranslateStringIn:
|
|
args = {}
|
|
for (k, v) in kwargs.items():
|
|
if isinstance(v, str):
|
|
args[k] = pb.TranslateArgValue(str=v)
|
|
else:
|
|
args[k] = pb.TranslateArgValue(number=v)
|
|
return pb.TranslateStringIn(key=key, args=args)
|
|
|
|
|
|
class Translations(GeneratedTranslations):
|
|
def __init__(self, backend: Optional[ref[RustBackend]]):
|
|
self.backend = backend
|
|
|
|
def __call__(self, *args: Any, **kwargs: Any) -> str:
|
|
"Mimic the old col.tr / TR interface"
|
|
return self.backend().translate(*args, **kwargs)
|
|
|
|
def _translate(
|
|
self, module: int, translation: int, args: Dict[str, Union[str, int, float]]
|
|
) -> str:
|
|
return self.backend().translate(module * 1000 + translation, **args)
|