112 lines
4 KiB
Python
112 lines
4 KiB
Python
"""Command line helpers.
|
|
|
|
python -m app.cli import [--replace] [--trainee-email EMAIL]
|
|
|
|
Imports all trainings from the Obsidian and Apple notes into the database as trainings of one
|
|
trainee. The trainer from a note header is matched with a user of the trainer role by
|
|
"First Last" ("соло" means no trainer). The schema must already exist (`alembic upgrade head`).
|
|
"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import logging
|
|
from typing import List, Optional, Set
|
|
|
|
from app.config import settings
|
|
from app.core.database.connection import create_engine, create_session_factory
|
|
from app.core.database.repository import TrainingRepository
|
|
from app.core.database.user_repository import UserRepository
|
|
from app.core.dto.training import TrainingDTO
|
|
from app.core.dto.user import UserRole
|
|
from app.core.parsers.apple import AppleNotesParser
|
|
from app.core.parsers.obsidian import ObsidianNotesParser
|
|
|
|
|
|
SOLO_NAMES = {"solo", "соло"}
|
|
|
|
|
|
async def assign_participants(
|
|
session, trainings: List[TrainingDTO], trainee_email: Optional[str]
|
|
) -> None:
|
|
"""Set trainee_id and trainer_id on the parsed trainings."""
|
|
users = await UserRepository(session).list()
|
|
trainees = [user for user in users if user.role == UserRole.trainee]
|
|
if trainee_email:
|
|
matching = [u for u in trainees if u.email == trainee_email.strip().lower()]
|
|
if not matching:
|
|
raise SystemExit(f"No trainee with email {trainee_email}")
|
|
trainee = matching[0]
|
|
elif len(trainees) == 1:
|
|
trainee = trainees[0]
|
|
else:
|
|
raise SystemExit("There is more than one trainee, pass --trainee-email")
|
|
|
|
trainer_ids = {
|
|
f"{user.first_name} {user.last_name}".strip().casefold(): user.id
|
|
for user in users
|
|
if user.role == UserRole.trainer
|
|
}
|
|
unknown: Set[str] = set()
|
|
for training in trainings:
|
|
training.trainee_id = trainee.id
|
|
name = (training.trainer or "").strip().casefold()
|
|
if name in SOLO_NAMES or not name:
|
|
training.trainer_id = None
|
|
elif name in trainer_ids:
|
|
training.trainer_id = trainer_ids[name]
|
|
else:
|
|
unknown.add(training.trainer or "")
|
|
if unknown:
|
|
raise SystemExit(f"No trainer users for: {', '.join(sorted(unknown))}")
|
|
|
|
|
|
async def import_notes(replace: bool, trainee_email: Optional[str]) -> None:
|
|
parsers = [ObsidianNotesParser(), AppleNotesParser()]
|
|
trainings = []
|
|
for parser in parsers:
|
|
parsed = parser.parse_file()
|
|
print(
|
|
f"{type(parser).__name__}: {len(parsed)} trainings, "
|
|
f"{len(parser.errors)} skipped lines"
|
|
)
|
|
trainings.extend(parsed)
|
|
|
|
engine = create_engine(settings.get_postgres_database_url())
|
|
try:
|
|
async with create_session_factory(engine)() as session:
|
|
repository = TrainingRepository(session)
|
|
await assign_participants(session, trainings, trainee_email)
|
|
existing = await repository.count()
|
|
if existing and not replace:
|
|
raise SystemExit(
|
|
f"Database already has {existing} trainings, "
|
|
"use --replace to overwrite them"
|
|
)
|
|
if replace:
|
|
await repository.delete_all()
|
|
await repository.save_many(trainings)
|
|
finally:
|
|
await engine.dispose()
|
|
print(f"Imported {len(trainings)} trainings")
|
|
|
|
|
|
def main() -> None:
|
|
logging.basicConfig(level=logging.WARNING)
|
|
arg_parser = argparse.ArgumentParser(prog="app.cli")
|
|
subcommands = arg_parser.add_subparsers(dest="command", required=True)
|
|
import_parser = subcommands.add_parser("import", help="import notes into the DB")
|
|
import_parser.add_argument(
|
|
"--replace", action="store_true", help="delete existing trainings first"
|
|
)
|
|
import_parser.add_argument(
|
|
"--trainee-email",
|
|
help="whose trainings these are (not needed when there is a single trainee)",
|
|
)
|
|
args = arg_parser.parse_args()
|
|
|
|
if args.command == "import":
|
|
asyncio.run(import_notes(args.replace, args.trainee_email))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|