m365-brain -- the one console script.
One entry point, not two. A two-letter alias on every user's PATH buys them a
shell alias they can write themselves and costs a second name to document and
keep in step.
--config is optional at the group level and required everywhere else.
init creates the config file and therefore cannot demand it already exists;
every other verb calls require_config, which fails with exit 3. Making it
optional at the group and enforcing it per-verb is the only arrangement in
which both are true.
Exit codes are defined once, as the EXIT_* constants in
commands/_context.py, each carrying the reasoning for its own existence.
ExitCodeGroup below maps the exception families onto them in one place rather
than in fifteen try blocks. CONTRACTS.md ยง Exit codes is the same table for
a reader who is not in the source; a third copy here was one more thing to keep
in step, and this file is where it would have been noticed last.
module-attribute
TEMPLATE_ANCHORS = ('../vault', '../state', '../notes', '../attachments')
The four directory prefixes the packaged template writes its paths under.
init rewrites each to somewhere under --vault. Literal string replacement
rather than a YAML round-trip, because the template's comments are the
configuration reference -- yaml.safe_dump would silently delete the document
that explains what the operator just got.
Bases: Group
Maps the failure families onto their exit codes, once.
NotFound is separate from ConfigError because they are separate facts.
Routing both through one class made index context nope exit 3 -- "your
configuration is invalid" -- when the config was fine and the corpus simply
had no match, while catalog search nope exited 0 for the same situation.
Source code in m365_brain/cli.py
| class ExitCodeGroup(click.Group):
"""Maps the failure families onto their exit codes, once.
`NotFound` is separate from `ConfigError` because they are separate facts.
Routing both through one class made `index context nope` exit 3 -- "your
configuration is invalid" -- when the config was fine and the corpus simply
had no match, while `catalog search nope` exited 0 for the same situation.
"""
def invoke(self, ctx: click.Context) -> object:
try:
return super().invoke(ctx)
except NotFound as exc:
click.echo(f"not found: {exc}", err=True)
raise SystemExit(EXIT_NOT_FOUND) from exc
except (ConfigError, HookResolutionError, VaultPathError) as exc:
click.echo(f"config error: {exc}", err=True)
raise SystemExit(EXIT_CONFIG) from exc
except (AuthProfileError, AuthRequiredError, TokenRefreshError) as exc:
click.echo(f"authentication required: {exc}", err=True)
raise SystemExit(EXIT_AUTH) from exc
|
main(ctx: Context, config_path: str | None) -> None
Sync Microsoft 365 into a markdown vault, index it, and write back.
Source code in m365_brain/cli.py
| @click.group(cls=ExitCodeGroup)
@click.option("--config", "config_path", type=str, default=None, help="Comma-separated config YAML paths")
@click.pass_context
def main(ctx: click.Context, config_path: str | None) -> None:
"""Sync Microsoft 365 into a markdown vault, index it, and write back."""
route_logs_to_stderr()
ctx.ensure_object(dict)
ctx.obj[CONFIG_KEY] = config_path
if config_path:
first = Path(config_path.split(",")[0].strip()).resolve().parent / ".env"
if first.exists():
load_dotenv(first)
load_dotenv(find_dotenv(usecwd=True))
|
init(path: Path, vault_dir: Path) -> None
Write a complete config file and create the vault directories.
Refuses to overwrite: a config file is the one artifact an operator has
edited by hand, and clobbering it to save a flag is not a trade.
Paths are written absolute. Relative ones would resolve against the config
file's directory, which is correct but invisible -- and init is the one
moment where the operator has not yet learned that rule.
Source code in m365_brain/cli.py
| @main.command("init")
@click.argument("path", type=click.Path(dir_okay=False, path_type=Path))
@click.option("--vault", "vault_dir", required=True, type=click.Path(file_okay=False, path_type=Path))
def init(path: Path, vault_dir: Path) -> None:
"""Write a complete config file and create the vault directories.
Refuses to overwrite: a config file is the one artifact an operator has
edited by hand, and clobbering it to save a flag is not a trade.
Paths are written absolute. Relative ones would resolve against the config
file's directory, which is correct but invisible -- and `init` is the one
moment where the operator has not yet learned that rule.
"""
if path.exists():
raise ConfigError(f"{path} already exists; delete it or choose another path")
resolved_vault = vault_dir.resolve()
template = resources.files(TEMPLATE_PACKAGE).joinpath(TEMPLATE_NAME)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(_relocate(template.read_text(encoding="utf-8"), resolved_vault), encoding="utf-8")
created = [str(path)]
for name in ("inbox", "annotations", "outbox", "attachments", "_meta"):
directory = resolved_vault / name
directory.mkdir(parents=True, exist_ok=True)
created.append(str(directory))
for line in created:
click.echo(line)
|
run(ctx: Context, once: bool, only: str | None, resync: bool, delay_start: int, as_json: bool) -> None
Run cycles: extract, index, then dispatch the post-cycle hooks.
--once changes two things, not one: a single cycle, and that cycle
ignoring poll_interval_minutes. Its help said "One cycle, then exit", so
the operator moving from a supervisor to cron -- the migration the flag
reads as supporting -- silently got every unit on every invocation.
Source code in m365_brain/cli.py
| @main.command("run")
@click.option("--once", "once", is_flag=True, help="One cycle, then exit -- every selected unit, due or not")
@click.option("--only", type=str, default=None, help="Comma-separated unit names")
@click.option("--resync", is_flag=True, help="Forget the selected extractors' delta tokens first")
@click.option("--delay-start", type=int, default=0, help="Minutes to wait before the first cycle", show_default=True)
@click.option("--json", "as_json", is_flag=True, help="Emit the manifest as JSON")
@click.pass_context
def run(ctx: click.Context, once: bool, only: str | None, resync: bool, delay_start: int, as_json: bool) -> None:
"""Run cycles: extract, index, then dispatch the post-cycle hooks.
`--once` changes two things, not one: a single cycle, and that cycle
ignoring `poll_interval_minutes`. Its help said "One cycle, then exit", so
the operator moving from a supervisor to cron -- the migration the flag
reads as supporting -- silently got every unit on every invocation.
"""
config = require_config(ctx)
selection = Selection(names=comma_list(only), resync=resync, ignore_schedule=once)
select_units(config, selection.names) # exit 3 before anything is built
runtime = build_runtime(config)
if not once:
raise SystemExit(run_forever(runtime, selection, delay_start * SECONDS_PER_MINUTE))
manifest = run_once(runtime, selection)
emit(as_json, manifest.model_dump(mode="json"), _cycle_lines(manifest))
if not manifest.ok:
raise SystemExit(EXIT_FAILURE)
|
extract(ctx: Context, only: str | None, resync: bool, dry: bool, as_json: bool) -> None
Run the extractors once, without the index step or the hooks.
A dry run has no manifest to emit, so --json beside it leaves stdout
empty. The two flags were advertised side by side with nothing saying so.
Source code in m365_brain/cli.py
| @main.command("extract")
@click.option("--only", type=str, default=None, help="Comma-separated extractor names")
@click.option("--resync", is_flag=True, help="Forget the selected extractors' delta tokens first")
@click.option("--dry-run", "dry", is_flag=True, help="Probe each endpoint, write nothing. Reports on stderr")
@click.option("--json", "as_json", is_flag=True, help="Emit the manifest as JSON. Not honoured under --dry-run")
@click.pass_context
def extract(ctx: click.Context, only: str | None, resync: bool, dry: bool, as_json: bool) -> None:
"""Run the extractors once, without the index step or the hooks.
A dry run has no manifest to emit, so `--json` beside it leaves stdout
empty. The two flags were advertised side by side with nothing saying so.
"""
from m365_brain.commands._context import token_provider
from m365_brain.config import EXTRACTOR_NAMES
from m365_brain.dry_run import dry_run
config = require_config(ctx)
names = comma_list(only)
if dry:
chosen = names or [name for name in EXTRACTOR_NAMES if getattr(config.extractors, name).enabled]
dry_run(config, token_provider(config), chosen)
return
extractors = names or [name for name in EXTRACTOR_NAMES if getattr(config.extractors, name).enabled]
select_units(config, extractors) # exit 3 before anything is built
manifest = run_once(build_runtime(config), Selection(names=extractors, resync=resync, ignore_schedule=True))
payload = {entry.name: entry.item_count for entry in manifest.extractors}
emit(as_json, manifest.model_dump(mode="json"), [f"{name}\t{count} item(s)" for name, count in payload.items()])
if not manifest.ok:
raise SystemExit(EXIT_FAILURE)
|
status(ctx: Context, as_json: bool) -> None
Per-unit last run, last success, failure streak, and the last cycle.
Source code in m365_brain/cli.py
| @main.command("status")
@click.option("--json", "as_json", is_flag=True, help="Emit JSON on stdout")
@click.pass_context
def status(ctx: click.Context, as_json: bool) -> None:
"""Per-unit last run, last success, failure streak, and the last cycle."""
from m365_brain.commands._context import state_store
from m365_brain.cycle import select_units
from m365_brain.manifest import ManifestStore
from m365_brain.vault.paths import manifest_directory
config = require_config(ctx)
state = state_store(config)
units = select_units(config, None)
cursors = {unit.name: read_cursor(state, unit.name) for unit in units}
vault = require_section(config.vault, "vault")
latest = ManifestStore(manifest_directory(vault), require_section(config.manifest, "manifest")).latest()
payload = {
"units": cursors,
"vault_root": str(VaultPaths(vault).vault.root),
"last_cycle": None
if latest is None
else {
"cycle_id": latest.cycle_id,
"finished_at": latest.finished_at.isoformat(),
"ok": latest.ok,
"failures": latest.failures(),
"hooks": [{"spec": hook.spec, "error": hook.error} for hook in latest.hooks],
},
}
emit(as_json, payload, _status_lines(cursors, latest))
unhealthy = any(int(cursor.get("consecutive_failures", 0)) for cursor in cursors.values())
if unhealthy or (latest is not None and not latest.ok):
raise SystemExit(EXIT_FAILURE)
|