256 lines
8.0 KiB
Python
256 lines
8.0 KiB
Python
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()
|