import abc import datetime import json import logging import re import sys import typing from abc import abstractmethod from concurrent.futures import ThreadPoolExecutor from typing import List import httpx import typer from typer import Typer, Option import sentry_sdk sentry_sdk.init( "https://915a1bc84eb94d34b36cfdef13b3f3e5@o271257.ingest.sentry.io/6040098", # Set traces_sample_rate to 1.0 to capture 100% # of transactions for performance monitoring. # We recommend adjusting this value in production. traces_sample_rate=1.0, ) logging.basicConfig(level=logging.DEBUG) class CookieStorage(abc.ABC): @abstractmethod def save_cookies(self, cookies: dict): raise NotImplemented @abstractmethod def load_cookies(self) -> typing.Optional[dict]: raise NotImplemented class FilesystemCookieStorage(CookieStorage): def __init__(self, filename: str): self.filename = filename def save_cookies(self, cookies: dict): with open(self.filename, "wt") as f: json.dump({"date": datetime.datetime.now().isoformat(), "cookies": cookies}, f) def load_cookies(self) -> typing.Optional[dict]: try: with open(self.filename) as f: saved = json.load(f) except FileNotFoundError: return None date_saved = datetime.datetime.fromisoformat(saved["date"]) if (datetime.datetime.now() - date_saved).days > 1: return None return saved["cookies"] class AnkiClient: def __init__(self, cookie_storage: CookieStorage = None): self._cookie_storage = cookie_storage or FilesystemCookieStorage("anki.json") self._logger = logging.getLogger(self.__class__.__name__) self._http = httpx.Client( headers={ "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/97.0.4681.0 Safari/537.36", "X-Requested-With": "XMLHttpRequest", }, verify=False, follow_redirects=False, ) def _load_cookies(self): cookies = self._cookie_storage.load_cookies() if not cookies: raise PermissionError("please login") self._http.cookies = cookies @property def session(self) -> dict: """ Returns session cookies :return: session cookies """ return self._cookie_storage.load_cookies() def login(self, username: str, password: str) -> dict: """ login and return cookies :return: cookies """ self._logger.debug("visiting login page") res = self._http.get("https://ankiweb.net/account/login", follow_redirects=True) res.raise_for_status() self._logger.debug("submitting login form") hidden_fields = dict(re.findall(r'.*type="hidden" name="([^"]+)" value="([^"]+)".*', res.text, re.MULTILINE)) res = self._http.post( "https://ankiweb.net/account/login", data={**hidden_fields, "username": username, "password": password}, follow_redirects=True ) res.raise_for_status() cookies = {**self._http.cookies} self._cookie_storage.save_cookies(cookies) return cookies def get_editor_context(self): self._load_cookies() add_info = self._http.get("https://ankiuser.net/edit/getAddInfo", follow_redirects=True).json() note_types = {it["name"]: {"id": it["id"], "name": it["name"]} for it in add_info["notetypes"]} decks = {it["name"]: {"id": it["id"], "name": it["name"]} for it in add_info["decks"]} with ThreadPoolExecutor(max_workers=len(note_types)) as pool: def fetch_note_fields(nid): res = self._http.get( "https://ankiuser.net/edit/getNotetypeFields", params={"ntid": nid}, follow_redirects=True, ) if res.is_error: return [] fields_info = res.json() return [f["name"] for f in fields_info["fields"]] results = dict( zip(note_types.keys(), pool.map(fetch_note_fields, [it["id"] for it in note_types.values()])) ) for t in note_types: note_types[t]["fields"] = results[t] return note_types, decks def create_note(self, note_type: str, deck: str, fields: dict, tags: str = ""): self._load_cookies() self._logger.debug("fetching note types") add_info = self._http.get("https://ankiuser.net/edit/getAddInfo").json() note_types = {it["name"]: it["id"] for it in add_info["notetypes"]} decks = {it["name"]: it["id"] for it in add_info["decks"]} self._logger.debug("fetching note fields") fields_info = self._http.get( "https://ankiuser.net/edit/getNotetypeFields", params={"ntid": note_types[note_type]} ).json() model_fields = [f["name"] for f in fields_info["fields"]] editor_html = self._http.get("https://ankiuser.net/edit/").text m = re.search(r"""new anki.Editor\('(ey.*)',.*\)""", editor_html) csrf_token = m.group(1) model_id = note_types[note_type] self._logger.debug(f"found note type: {model_id}") self._logger.debug(f"note fields: {model_fields}") # unused fields must be an empty string, otherwise we get a 403 ordered_fields = [fields.get(f, "") for f in model_fields] payload = { "nid": "", "data": json.dumps([ordered_fields, tags]), "csrf_token": csrf_token, "mid": model_id, "deck": decks[deck], } self._logger.debug("saving note") res = self._http.post("https://ankiuser.net/edit/save", data=payload) res.raise_for_status() app = Typer(name="typer") state = {"session": "anki.json", "creds": "anki.creds.json"} @app.callback() def callback(session: str = "anki.json", creds: str = "anki.creds.json"): state.update(session=session, creds=creds) def make_anki() -> AnkiClient: anki = AnkiClient(FilesystemCookieStorage(state["session"])) if not anki.session: typer.echo("no valid session found, please login", err=True) raise typer.Exit(1) return anki def try_login(): try: with open(state["creds"]) as f: creds = json.load(f) anki_login(username=creds["username"], password=creds["password"]) except FileNotFoundError: pass @app.command("login") def anki_login( username: str = Option(..., prompt=True), password: str = Option(..., prompt=True), force: bool = False, ): with open(state["creds"], "wt") as f: json.dump({"username": username, "password": password}, f) anki = AnkiClient(FilesystemCookieStorage(state["session"])) if force or not anki.session: anki.login(username=username, password=password) @app.command("info") def anki_info(): try_login() anki = make_anki() note_types, decks = anki.get_editor_context() logging.info("note types:") for _, t in note_types.items(): logging.info(t) logging.info("decks:") for d in decks.items(): logging.info(d) @app.command("note") def anki_create_note( deck: str = Option(...), type: str = Option(...), fields: List[str] = Option(..., "-f", "--field", help="Fields in k=v format"), tags: str = "", ): try_login() anki = make_anki() parsed_fields = dict([it.split("=", maxsplit=1) for it in fields]) anki.create_note( deck=deck, note_type=type, fields=parsed_fields, tags=tags, ) if __name__ == "__main__": logging.debug(f"Called with {sys.argv=}") # try_login() # anki = make_anki() # anki.create_note( # deck="Deutsch", # note_type="Card with Reverse and Examples", # fields={ # "Front": "Es hat gerade an der Tür aufgeklingelt. Kannst du aufmachen?", # "Back": "The doorbell just rang. Can you get it?", # "Examples": "", # }, # ) app()