mews.page/mews/cli.py

"""mewsd: run the service, and look after the directory from the shell."""

import argparse
import sys
import time

from mews import check as checker, db, pages
from mews.app import directory_path, serve
from mews.fetch import Fetcher, registered_domain

RECHECK_BATCH = 50
PAUSE_SECONDS = 2.0


def main(argv: list[str] | None = None) -> int:
    """Run one command. Returns an exit status."""
    parser = argparse.ArgumentParser(
        prog="mewsd", description="Run and look after the mews.page directory."
    )
    commands = parser.add_subparsers(dest="command", required=True)
    commands.add_parser("serve", help="serve the check endpoint")
    commands.add_parser("init-db", help="create the database if it isn't there")
    commands.add_parser("recheck", help="check the sites that are due")
    commands.add_parser("build-directory", help="write the directory page again")
    drop = commands.add_parser("remove", help="take a site out and refuse it later")
    drop.add_argument("domain")
    drop.add_argument("--reason", required=True, help="shown in the log and to you")
    allow = commands.add_parser("unblock", help="allow a removed domain again")
    allow.add_argument("domain")
    show = commands.add_parser("show", help="what the last check of a site found")
    show.add_argument("domain")
    args = parser.parse_args(argv)

    if args.command == "serve":
        serve()
        return 0

    connection = db.connect()
    try:
        db.init(connection)
        if args.command == "init-db":
            print(f"database ready at {db.path()}")
            return 0
        if args.command == "recheck":
            return _recheck(connection)
        if args.command == "build-directory":
            return _build(connection)
        if args.command == "remove":
            return _remove(connection, args.domain, args.reason)
        if args.command == "unblock":
            return _unblock(connection, args.domain)
        if args.command == "show":
            return _show(connection, args.domain)
    finally:
        connection.close()
    return 0


def _domain(name: str) -> str:
    return registered_domain(name) or name.strip().lower()


def _recheck(connection) -> int:
    """Check every site that is due, then write the directory page again."""
    sites = db.due(connection, RECHECK_BATCH)
    counts = {"pass": 0, "fail": 0, "unreachable": 0}
    for index, row in enumerate(sites):
        if index:
            # One site at a time, with a gap, so a recheck pass is invisible to
            # the sites it visits.
            time.sleep(PAUSE_SECONDS)
        with Fetcher() as fetcher:
            counts[checker.recheck(connection, row, fetcher=fetcher)] += 1
    db.prune(connection)
    _build(connection)
    print(
        f"checked {len(sites)}: {counts['pass']} passed, {counts['fail']} dropped, "
        f"{counts['unreachable']} unreachable"
    )
    return 0


def _build(connection) -> int:
    """Write the directory page from what is listed now."""
    try:
        warnings = pages.write_directory(directory_path(), db.listed(connection))
    except ValueError as error:
        print(f"the directory page wasn't written: {error}", file=sys.stderr)
        return 1
    for warning in warnings:
        print(f"note: {warning}", file=sys.stderr)
    print(f"wrote {directory_path()}")
    return 0


def _remove(connection, domain: str, reason: str) -> int:
    name = _domain(domain)
    row = db.site(connection, name)
    if row is not None:
        db.remove(connection, row["id"], reason)
    db.block(connection, name, reason)
    _build(connection)
    print(f"removed {name}")
    return 0


def _unblock(connection, domain: str) -> int:
    name = _domain(domain)
    db.unblock(connection, name)
    print(f"{name} can be submitted again")
    return 0


def _show(connection, domain: str) -> int:
    name = _domain(domain)
    row = db.site(connection, name)
    if row is None:
        print(f"{name} isn't in the directory", file=sys.stderr)
        return 1
    print(f"{row['domain']} — {row['state']}")
    print(f"  page      {row['url']}")
    print(f"  title     {row['title']}")
    print(f"  listed    {row['listed_at']}")
    print(f"  checked   {row['last_check_at']} (last passed {row['last_ok_at']})")
    print(f"  next      {row['next_check_at']}")
    if row["reason"]:
        print(f"  dropped   {row['reason']}")
    findings = db.last_findings(connection, row["id"])
    for finding in findings:
        print(
            f"  {finding['level']:6} section {finding['section']} — "
            f"{finding['message']}"
        )
    return 0


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