maa-online/bin/maa-resume-tasks.py

#!/usr/bin/env python3
"""Build a resume task file from partially finished MAA daily attempts.

A failed daily attempt still completes some tasks (their progress lives on the
game server), so re-running the whole daily file on a retry wastes cloud-game
minutes. This helper reads the previous attempts' MAA logs, finds which task
types already reported "<Type> Completed", and writes a task file containing
only the remaining tasks. The daily runner (bin/maa-online-daily-run) calls it
automatically before every retry.

Usage:
    maa-resume-tasks.py <log> [<log> ...] [--source <daily.toml>] [--out <file>]

Output (stdout, last line):
    RESUME_TASKS=<name1>,<name2>   remaining tasks written to --out
    NONE                           nothing left to do
    ERROR:<reason>                 could not build a resume file (caller should
                                   fall back to the full daily run)

StartUp is intentionally never skipped: every /start yields a fresh cloud
instance, and the global `startup = true` flag guarantees the game is launched
and sitting on the main screen before the remaining tasks run.
"""

from __future__ import annotations

import argparse
import os
import re
import sys
from pathlib import Path

ROOT = Path(os.environ.get("MAA_ONLINE_ROOT") or Path(__file__).resolve().parent.parent)
DEFAULT_SOURCE = Path(os.environ.get("MAA_ONLINE_DAILY_SOURCE", ROOT / "config/tasks/daily.toml"))
DEFAULT_OUT = Path(os.environ.get("MAA_ONLINE_RESUME_OUT", ROOT / "config/tasks/daily-resume.toml"))

TASK_HEADER = re.compile(r"^\[\[tasks\]\]\s*$")
TYPE_RE = re.compile(r'^type\s*=\s*"([^"]+)"\s*$', re.MULTILINE)
NAME_RE = re.compile(r'^name\s*=\s*"([^"]+)"\s*$', re.MULTILINE)
COMPLETED_RE = re.compile(r"^.*\b(\w+) Completed\s*$", re.MULTILINE)

# Task types that are never skipped even when a completion marker is present:
# a retry always reconnects to a fresh cloud instance and must relaunch/verify
# the game before the remaining tasks can run.
NEVER_SKIP = {"StartUp"}


def split_source(text: str) -> tuple[str, list[str]]:
    """Return (prelude, task blocks).

    A block owns the contiguous comment lines directly above its [[tasks]]
    header, so a "# 1. xxx" section comment travels with its task instead of
    being left behind in the prelude. Blocks keep their exact formatting.
    """
    lines = text.splitlines(keepends=True)
    header_indexes = [i for i, line in enumerate(lines) if TASK_HEADER.match(line)]
    if not header_indexes:
        raise ValueError("no [[tasks]] blocks found in source file")

    starts = []
    for index in header_indexes:
        start = index
        while start - 1 >= 0 and lines[start - 1].lstrip().startswith("#"):
            start -= 1
        starts.append(start)

    prelude = "".join(lines[: starts[0]])
    blocks = []
    for position, start in enumerate(starts):
        end = starts[position + 1] if position + 1 < len(starts) else len(lines)
        blocks.append("".join(lines[start:end]).strip("\n") + "\n")
    return prelude, blocks


def completed_types(log_text: str) -> set[str]:
    return set(COMPLETED_RE.findall(log_text))


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("logs", nargs="+", help="previous attempts' MAA log files")
    parser.add_argument("--source", default=str(DEFAULT_SOURCE))
    parser.add_argument("--out", default=str(DEFAULT_OUT))
    args = parser.parse_args()

    source = Path(args.source)
    if not source.is_file():
        print(f"ERROR:source task file not found: {source}")
        return 0

    try:
        prelude, blocks = split_source(source.read_text(encoding="utf-8"))
    except (OSError, ValueError) as exc:
        print(f"ERROR:{exc}")
        return 0

    done: set[str] = set()
    for log in args.logs:
        log_path = Path(log)
        if not log_path.is_file():
            continue
        try:
            done |= completed_types(log_path.read_text(encoding="utf-8", errors="replace"))
        except OSError as exc:
            print(f"ERROR:cannot read previous log {log}: {exc}")
            return 0

    kept: list[str] = []
    kept_names: list[str] = []
    for block in blocks:
        match = TYPE_RE.search(block)
        if not match:
            continue
        task_type = match.group(1)
        if task_type in done and task_type not in NEVER_SKIP:
            continue
        kept.append(block)
        name_match = NAME_RE.search(block)
        kept_names.append(name_match.group(1) if name_match else task_type)

    if not kept:
        print("NONE")
        return 0

    header = (
        "# AUTO-GENERATED by bin/maa-resume-tasks.py -- do not edit.\n"
        f"# Source: {source}\n"
        f"# Already completed (skipped): {', '.join(sorted(done)) or 'none'}\n\n"
    )
    out = Path(args.out)
    try:
        out.parent.mkdir(parents=True, exist_ok=True)
        out.write_text(
            header + prelude.rstrip("\n") + "\n\n" + "\n".join(kept), encoding="utf-8"
        )
    except OSError as exc:
        print(f"ERROR:cannot write resume file: {exc}")
        return 0

    print(f"RESUME_TASKS={','.join(kept_names)}")
    return 0


if __name__ == "__main__":
    sys.exit(main())