Newer
Older
thu-learn-downloader-optimized / thu_learn_downloader / sync.py
@Qin Li Qin Li on 22 Mar 2023 7 KB fix: migrate to typer
import os
import shutil
import subprocess
from datetime import datetime
from pathlib import Path
from typing import Optional

from rich.console import Console
from rich.progress import Progress, TaskID

from . import typing as t
from . import utils
from .config import Config
from .constants import DOCUMENT_STYLE, HOMEWORK_STYLE, SKIPPED_PREFIX, SUCCESS_PREFIX
from .downloader import Downloader
from .helper import Helper


def sync_all(
    helper: Helper,
    downloader: Downloader,
    config: Config,
    *,
    console: Console = Console(),
    overall_progress: Progress = Progress(),
    semesters_task_id: TaskID = TaskID(0),
    courses_task_id: TaskID = TaskID(1),
) -> None:
    semesters = config.semesters or helper.get_semester_id_list()

    for semester in overall_progress.track(
        semesters, task_id=semesters_task_id, description="Semesters"
    ):
        sync_semester(
            helper=helper,
            downloader=downloader,
            semester_id=semester,
            config=config,
            console=console,
            overall_progress=overall_progress,
            courses_task_id=courses_task_id,
        )


def sync_semester(
    helper: Helper,
    downloader: Downloader,
    config: Config,
    semester_id: str,
    *,
    console: Console = Console(),
    overall_progress: Progress = Progress(),
    courses_task_id: TaskID = TaskID(1),
) -> None:
    courses: list[t.CourseInfo] = helper.get_course_list(semester_id=semester_id)

    if config.courses:
        courses = [
            course
            for course in courses
            if (course.name in config.courses)
            or (course.english_name in config.courses)
        ]

    overall_progress.update(
        task_id=courses_task_id, description=utils.format_semester_id(semester_id)
    )
    for course in overall_progress.track(courses, task_id=courses_task_id):
        sync_course(
            helper=helper,
            downloader=downloader,
            config=config,
            course=course,
            console=console,
        )


def sync_course(
    helper: Helper,
    downloader: Downloader,
    config: Config,
    course: t.CourseInfo,
    *,
    console: Console = Console(),
) -> None:
    sync_files(
        helper=helper,
        downloader=downloader,
        config=config,
        course=course,
        console=console,
    )
    sync_homeworks(
        helper=helper,
        downloader=downloader,
        config=config,
        course=course,
        console=console,
    )


def sync_files(
    helper: Helper,
    downloader: Downloader,
    config: Config,
    course: t.CourseInfo,
    *,
    console: Console = Console(),
) -> None:
    prefix: Path = config.prefix
    size_limit: int = config.size_limit

    files: list[t.File] = helper.get_file_list(course_id=course.id)
    files: list[t.File] = sorted(files, key=lambda f: f.id)

    for i, f in enumerate(files):
        filename: str = utils.format_doc_filename(title=f.title, file_type=f.file_type)
        filename: str = f"{i:02d}-{filename}"
        if f.raw_size > size_limit:
            console.log(
                SKIPPED_PREFIX,
                utils.describe_doc_file(
                    course_name=course.english_name, filename=filename
                ),
                "size limit exceeded",
                style=DOCUMENT_STYLE,
            )
            continue

        downloader.schedule_download(
            url=f.download_url,
            output=prefix / course.english_name / "docs" / f.file_clazz / filename,
            session=helper,
            raw_size=f.raw_size,
            upload_time=f.upload_time,
            console=console,
            style=DOCUMENT_STYLE,
            description=utils.describe_doc_file(
                course_name=course.english_name, filename=filename
            ),
        )


def sync_homeworks(
    helper: Helper,
    downloader: Downloader,
    config: Config,
    course: t.CourseInfo,
    *,
    console: Console = Console(),
) -> None:
    homeworks = helper.get_homework_list(course_id=course.id)

    for hw in homeworks:
        sync_homework_detail(
            helper=helper,
            downloader=downloader,
            config=config,
            course=course,
            hw=hw,
            console=console,
        )
        sync_homework_attachments(
            helper=helper,
            downloader=downloader,
            config=config,
            course=course,
            hw=hw,
            console=console,
        )


def sync_homework_detail(
    helper: Helper,
    downloader: Downloader,
    config: Config,
    course: t.CourseInfo,
    hw: t.Homework,
    *,
    console: Console = Console(),
) -> None:
    prefix: Path = config.prefix
    title: str = f"{hw.number:02d}-{hw.title}"
    filepath: Path = prefix / course.english_name / "work" / title / "README.md"
    os.makedirs(filepath.parent, exist_ok=True)
    with open(file=filepath, mode="w") as fp:
        fp.write(utils.format_homework_readme(hw=hw))
        fp.flush()
    if shutil.which("prettier"):
        subprocess.run(args=["prettier", "--write", filepath], capture_output=True)
    console.log(
        SUCCESS_PREFIX,
        course.english_name,
        ">",
        title,
        style=HOMEWORK_STYLE,
    )


def sync_homework_attachments(
    helper: Helper,
    downloader: Downloader,
    config: Config,
    course: t.CourseInfo,
    hw: t.Homework,
    *,
    console: Console = Console(),
) -> None:
    sync_homework_attachment(
        helper=helper,
        downloader=downloader,
        config=config,
        course=course,
        hw=hw,
        attach=hw.attachment,
        attach_type="attach",
        console=console,
    )
    sync_homework_attachment(
        helper=helper,
        downloader=downloader,
        config=config,
        course=course,
        hw=hw,
        attach=hw.answer_attachment,
        attach_type="ans",
        console=console,
    )
    sync_homework_attachment(
        helper=helper,
        downloader=downloader,
        config=config,
        course=course,
        hw=hw,
        attach=hw.submitted_attachment,
        attach_type="submit",
        console=console,
    )
    sync_homework_attachment(
        helper=helper,
        downloader=downloader,
        config=config,
        course=course,
        hw=hw,
        attach=hw.grade_attachment,
        attach_type="comment",
        console=console,
    )


def sync_homework_attachment(
    helper: Helper,
    downloader: Downloader,
    config: Config,
    course: t.CourseInfo,
    hw: t.Homework,
    attach: Optional[t.RemoteFile],
    attach_type: str,
    *,
    console: Console = Console(),
) -> None:
    if not attach:
        return
    prefix: Path = config.prefix
    title: str = f"{hw.number:02d}-{hw.title}"
    filename: str = utils.remove_attachment_prefix(attach.name)
    filename: str = f"{attach_type}-{filename}"
    filepath: Path = prefix / course.english_name / "work" / title / filename
    upload_time: Optional[datetime] = None
    match attach_type:
        case "attach":
            upload_time = hw.starts_time
        case "ans":
            pass
        case "submit":
            upload_time = hw.submit_time
        case "comment":
            upload_time = hw.grade_time
    downloader.schedule_download(
        url=attach.download_url,
        output=filepath,
        session=helper,
        upload_time=upload_time,
        console=console,
        style=HOMEWORK_STYLE,
        description=utils.describe_work_file(
            course_name=course.english_name, hw_title=title, filename=filename
        ),
    )