inbound-library/inbound_library/__main__.py

"""Command line for inbound-library. Run with python3 -m inbound_library."""
import argparse
import fcntl
import json
from datetime import datetime, timezone
import sys
from pathlib import Path

from . import intake, notify, store, vault
from . import registry as reg

DEFAULT_CONFIG = Path('config') / 'library.json'


def _config(root, path):
    path = Path(path) if path else Path(root) / DEFAULT_CONFIG
    return json.loads(path.read_text()) if path.exists() else {}


def _print(value):
    print(json.dumps(value, indent=2, ensure_ascii=False))


def run_due(root, config, mailbox=None, post=notify.post_message):
    """One timer wake: intake, re-route quarantine, match, notify, digests."""
    state = Path(root) / 'data' / 'state'
    state.mkdir(parents=True, exist_ok=True)
    with (state / 'run.lock').open('a') as lock:
        try:
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            return {'ok': True, 'status': 'already_running'}
        registry = reg.load(root)
        problems = reg.validate(registry)
        if problems:
            return {'ok': False, 'status': 'invalid_registry', 'problems': problems}
        report = {'ok': True, 'status': 'checked'}
        try:
            mailbox = mailbox or intake.mailbox_from_config(config)
            if mailbox is None:
                report['intake'] = {'ok': False, 'status': 'needs_credentials'}
            else:
                report['intake'] = {'ok': True,
                                    'retained': len(intake.scan(root, mailbox, registry))}
        except vault.CredentialUnavailable as exc:
            report['intake'] = {'ok': False, 'status': 'credential_unavailable', 'error': str(exc)}
        except (OSError, ValueError) as exc:
            report['intake'] = {'ok': False, 'status': 'error', 'error': str(exc)}
        report['ok'] = report['intake']['ok']
        store.atomic(state / 'last-intake.json', json.dumps(
            {**report['intake'], 'at': datetime.now(timezone.utc).isoformat()}, indent=2).encode() + b'\n')
        report['requeued'] = len(intake.requeue_quarantine(root, registry))
        report['notices'] = len(notify.match(root, registry))
        hub = config.get('hub', {})
        report['hub'] = notify.deliver_hub(root, hub.get('api_base', 'http://127.0.0.1:8000'),
                                           hub.get('from_agent', 'inbound-library'),
                                           registry, post)
        if any(not r['ok'] for r in report['hub'].values()):
            report['ok'] = False
        report['digests'] = notify.write_digests(root, registry=registry)
        return report


def main(argv=None):
    parser = argparse.ArgumentParser(
        prog='python3 -m inbound_library',
        description='Keep delivered material and route it by topic to subscribing repos.')
    parser.add_argument('--root', default='.', help='repository root (default: .)')
    parser.add_argument('--config', help='config file (default: config/library.json)')
    sub = parser.add_subparsers(dest='command', required=True)
    sub.add_parser('check', help='validate registries and rehash every retained item')
    scan = sub.add_parser('scan', help='read the mailbox read-only and retain new mail')
    scan.add_argument('--fixture', help='scan a directory of .eml files instead of IMAP')
    sub.add_parser('run-due', help='intake, match, notify and write digests; safe to repeat')
    new = sub.add_parser('new', help='list items routed to a repo')
    new.add_argument('--repo', required=True)
    new.add_argument('--topic')
    new.add_argument('--since', help='ISO date or timestamp')
    show = sub.add_parser('show', help='show an item: derived text and deliveries')
    show.add_argument('item')
    get = sub.add_parser('get', help='write the original bytes of an item to stdout')
    get.add_argument('item')
    sub.add_parser('health', help='last scan, per-source counts, silence, quarantine')
    sources = sub.add_parser('sources', help='list registered sources a repo may see, by topic')
    sources.add_argument('--repo', required=True)
    sources.add_argument('--topic', help='limit to one topic (default: the repo subscription topics)')
    args = parser.parse_args(argv)
    root = Path(args.root)
    config = _config(root, args.config)

    if args.command == 'check':
        problems = reg.validate(reg.load(root)) + store.verify(root)
        for problem in problems:
            print(problem, file=sys.stderr)
        if problems:
            return 1
        print(f"Registry and store valid; {len(store.records(root, 'deliveries'))} deliveries")
        return 0
    if args.command == 'scan':
        try:
            mailbox = (intake.DirectoryMailbox(args.fixture) if args.fixture
                       else intake.mailbox_from_config(config))
        except vault.CredentialUnavailable as exc:
            print(f'credential_unavailable: {exc}', file=sys.stderr)
            return 2
        if mailbox is None:
            print('needs_credentials: configure "credential" in config/library.json '
                  'or set INBOUND_LIBRARY_IMAP_USER/PASSWORD', file=sys.stderr)
            return 2
        _print({'retained': intake.scan(root, mailbox)})
        return 0
    if args.command == 'run-due':
        report = run_due(root, config)
        _print(report)
        return 0 if report['ok'] else 1
    if args.command == 'new':
        _print(notify.notices_for(root, args.repo, args.topic, args.since))
        return 0
    if args.command == 'show':
        deliveries = [d for d in store.records(root, 'deliveries') if d['item'] == args.item]
        _print({'item': args.item, 'derived': store.get_derived(root, args.item),
                'deliveries': deliveries})
        return 0
    if args.command == 'get':
        sys.stdout.buffer.write(store.get_item(root, args.item))
        return 0
    if args.command == 'sources':
        _print(reg.sources_for(reg.load(root), args.repo, args.topic))
        return 0
    if args.command == 'health':
        report = notify.health(root)
        _print(report)
        return 0 if not report['integrity_problems'] else 1
    return 2


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