Skip to content

feat(cli): let every command write its result to a file #147

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
L4XB merged 1 commit into from
Sep 15, 2026
+347 −47
Merged

feat(cli): let every command write its result to a file #147

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,11 +12,16 @@ equivalents for Python package metadata.

- `sixsentences --version` and `six-community --version` report the installed
version of each command-line tool.
- Every `sixsentences` subcommand accepts `--output PATH`. A file receives
exactly the bytes stdout would have received, and is written only after the
command succeeds.

### Changed

- The server package reads its version from the installed distribution metadata
instead of a second literal in `sixsentences_server/__init__.py`.
- `sixsentences prisma --format svg --output PATH` now ends the file with a
newline, like every other written result.

## [0.2.0-alpha.1] - 2026-09-14

Expand Down
145 changes: 98 additions & 47 deletions src/sixsentences/cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,12 +6,12 @@
import json
import os
import sys
from collections.abc import Sequence
from collections.abc import Iterable, Sequence
from dataclasses import asdict
from pathlib import Path
from typing import Any, cast

from pydantic import ValidationError
from pydantic import BaseModel, ValidationError

from sixsentences import __version__
from sixsentences.connectors.openalex import OpenAlexClient, OpenAlexError
Expand Down Expand Up @@ -55,36 +55,53 @@ def _load_jsonl(path: Path) -> list[WorkRecord]:
return records


def _print_json(value: object) -> None:
def _json_text(value: object) -> str:
if hasattr(value, "model_dump_json"):
print(value.model_dump_json(indent=2))
return
print(json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True))
return cast(str, value.model_dump_json(indent=2)) + "\n"
return json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n"


def _jsonl_text(records: Iterable[BaseModel]) -> str:
return "".join(record.model_dump_json() + "\n" for record in records)


def _emit(args: argparse.Namespace, text: str) -> None:
"""Write one complete result to `--output`, or to stdout.

A file receives exactly the bytes stdout would have received, so redirection
and `--output` cannot disagree about a trailing newline. The payload is
built in full before the path is touched, so a command that fails partway
leaves no half-written artifact behind.
"""

def _emit_records(records: Sequence[WorkRecord]) -> None:
for record in records:
print(record.model_dump_json())
destination: Path | None = getattr(args, "output", None)
if destination is None:
sys.stdout.write(text)
return
destination.write_text(text, encoding="utf-8")


def _command_query(args: argparse.Namespace) -> None:
node = parse_query(args.query)
if args.target == "display":
print(to_display(node))
_emit(args, to_display(node) + "\n")
elif args.target == "openalex":
query, notes = compile_openalex(node)
_print_json(
{
"query": query,
"dropped_fields": notes.dropped_fields,
"dropped_wildcards": notes.dropped_wildcards,
}
_emit(
args,
_json_text(
{
"query": query,
"dropped_fields": notes.dropped_fields,
"dropped_wildcards": notes.dropped_wildcards,
}
),
)
elif args.target == "duckdb":
sql, parameters = compile_duckdb(node)
_print_json({"sql": sql, "parameters": parameters})
_emit(args, _json_text({"sql": sql, "parameters": parameters}))
else:
print(translations(node)[args.target])
_emit(args, translations(node)[args.target] + "\n")


def _command_corpus_build(args: argparse.Namespace) -> None:
Expand All @@ -93,13 +110,13 @@ def _command_corpus_build(args: argparse.Namespace) -> None:
args.corpus,
source=args.source,
)
_print_json(manifest)
_emit(args, _json_text(manifest))


def _command_corpus_search(args: argparse.Namespace) -> None:
corpus = LocalCorpus(args.corpus)
corpus.verify()
_emit_records(corpus.search(args.query, limit=args.limit))
_emit(args, _jsonl_text(corpus.search(args.query, limit=args.limit)))


def _command_openalex_search(args: argparse.Namespace) -> None:
Expand All @@ -116,7 +133,7 @@ def _command_openalex_search(args: argparse.Namespace) -> None:
)
finally:
client.close()
_emit_records(records)
_emit(args, _jsonl_text(records))


def _command_rank(args: argparse.Namespace) -> None:
Expand All @@ -126,51 +143,53 @@ def _command_rank(args: argparse.Namespace) -> None:
protocol,
now_year=args.now_year,
)
for item in ranked:
print(item.model_dump_json())
_emit(args, _jsonl_text(ranked))


def _command_coverage(args: argparse.Namespace) -> None:
payload = _load_json(args.captures)
if not isinstance(payload, dict):
raise ValueError("captures must be a JSON object mapping work ids to counts")
counts = cast(dict[str, int], payload)
_print_json(
estimate_completeness(
counts,
args.occasions,
occasion_independence_verified=args.occasion_independence_verified,
)
_emit(
args,
_json_text(
estimate_completeness(
counts,
args.occasions,
occasion_independence_verified=args.occasion_independence_verified,
)
),
)


def _command_prisma(args: argparse.Namespace) -> None:
counts = PrismaCounts.model_validate(_load_json(args.counts))
output = render_flow_svg(counts) if args.format == "svg" else render_flow_text(counts)
if args.output is None:
print(output)
else:
args.output.write_text(output + ("\n" if args.format == "text" else ""), encoding="utf-8")
rendered = render_flow_svg(counts) if args.format == "svg" else render_flow_text(counts)
_emit(args, rendered + "\n")


def _command_expansion_validate(args: argparse.Namespace) -> None:
payload = _load_json(args.candidates)
if not isinstance(payload, list) or not all(isinstance(item, str) for item in payload):
raise ValueError("candidates must be a JSON array of query strings")
_print_json(validate_variants(args.existing, payload, limit=args.limit))
_emit(args, _json_text(validate_variants(args.existing, payload, limit=args.limit)))


def _command_data_profile(args: argparse.Namespace) -> None:
dataset = parse_dataset_file(args.input)
_print_json(
{
"filename": dataset.filename,
"format": dataset.format,
"byte_count": dataset.byte_count,
"sha256": dataset.sha256,
"profile": asdict(dataset.profile),
"import_notes": dataset.import_notes,
}
_emit(
args,
_json_text(
{
"filename": dataset.filename,
"format": dataset.format,
"byte_count": dataset.byte_count,
"sha256": dataset.sha256,
"profile": asdict(dataset.profile),
"import_notes": dataset.import_notes,
}
),
)


Expand All @@ -197,7 +216,24 @@ def _command_data_analyze(args: argparse.Namespace) -> None:
)
else:
raise ValueError(f"unknown analysis recipe {args.analysis_kind!r}")
_print_json(asdict(analyze(dataset, recipe)))
_emit(args, _json_text(asdict(analyze(dataset, recipe))))


def _add_output(parser: argparse.ArgumentParser, result: str, *, nested: bool = False) -> None:
"""Give one command the shared `--output` option.

`nested` suppresses the default on a recipe parser, so that accepting the
option in both positions does not let the recipe overwrite a value the
parent already read.
"""

parser.add_argument(
"--output",
type=Path,
default=argparse.SUPPRESS if nested else None,
metavar="PATH",
help=f"write the {result} to PATH instead of stdout",
)


def _parser() -> argparse.ArgumentParser:
Expand All @@ -222,18 +258,21 @@ def _parser() -> argparse.ArgumentParser:
choices=("display", "openalex", "duckdb", "pubmed", "scopus", "wos", "ieee"),
default="display",
)
_add_output(query, "compiled query")
query.set_defaults(handler=_command_query)

corpus_build = subcommands.add_parser("corpus-build", help="build a local corpus from JSONL")
corpus_build.add_argument("input", type=Path)
corpus_build.add_argument("corpus", type=Path)
corpus_build.add_argument("--source", default="local-jsonl")
_add_output(corpus_build, "corpus manifest")
corpus_build.set_defaults(handler=_command_corpus_build)

corpus_search = subcommands.add_parser("corpus-search", help="search a local corpus")
corpus_search.add_argument("corpus", type=Path)
corpus_search.add_argument("query")
corpus_search.add_argument("--limit", type=int, default=100)
_add_output(corpus_search, "matching records")
corpus_search.set_defaults(handler=_command_corpus_search)

openalex = subcommands.add_parser(
Expand All @@ -245,12 +284,14 @@ def _parser() -> argparse.ArgumentParser:
openalex.add_argument("--limit", type=int, default=200)
openalex.add_argument("--year-from", type=int)
openalex.add_argument("--year-to", type=int)
_add_output(openalex, "fetched records")
openalex.set_defaults(handler=_command_openalex_search)

rank = subcommands.add_parser("rank", help="rank JSONL works against a protocol")
rank.add_argument("input", type=Path)
rank.add_argument("protocol", type=Path)
rank.add_argument("--now-year", type=int)
_add_output(rank, "ranked records")
rank.set_defaults(handler=_command_rank)

coverage = subcommands.add_parser("coverage", help="estimate search coverage with Chao2")
Expand All @@ -264,12 +305,13 @@ def _parser() -> argparse.ArgumentParser:
"without it the estimate is undetermined"
),
)
_add_output(coverage, "coverage estimate")
coverage.set_defaults(handler=_command_coverage)

prisma = subcommands.add_parser("prisma", help="render explicit PRISMA counters")
prisma.add_argument("counts", type=Path)
prisma.add_argument("--format", choices=("text", "svg"), default="text")
prisma.add_argument("--output", type=Path)
_add_output(prisma, "rendered flow")
prisma.set_defaults(handler=_command_prisma)

expansion = subcommands.add_parser(
Expand All @@ -278,26 +320,32 @@ def _parser() -> argparse.ArgumentParser:
expansion.add_argument("candidates", type=Path)
expansion.add_argument("--existing", action="append", default=[])
expansion.add_argument("--limit", type=int, default=5)
_add_output(expansion, "validation result")
expansion.set_defaults(handler=_command_expansion_validate)

data_profile = subcommands.add_parser(
"data-profile", help="profile a bounded CSV, TSV, JSON or XLSX dataset"
)
data_profile.add_argument("input", type=Path)
_add_output(data_profile, "dataset profile")
data_profile.set_defaults(handler=_command_data_profile)

data_analyze = subcommands.add_parser(
"data-analyze", help="run a deterministic analysis recipe over a bounded dataset"
)
data_analyze.add_argument("input", type=Path)
# Accepted before the recipe and after it, because both read naturally.
_add_output(data_analyze, "analysis result")
analysis_recipes = data_analyze.add_subparsers(dest="analysis_kind", required=True)

analysis_recipes.add_parser("missingness", help="count null and blank cells")
missingness = analysis_recipes.add_parser("missingness", help="count null and blank cells")
_add_output(missingness, "analysis result", nested=True)

descriptive = analysis_recipes.add_parser(
"descriptive", help="describe one complete numeric column"
)
descriptive.add_argument("--column", required=True)
_add_output(descriptive, "analysis result", nested=True)

group_summary = analysis_recipes.add_parser(
"group-summary", help="aggregate complete numeric values by group"
Expand All @@ -307,19 +355,22 @@ def _parser() -> argparse.ArgumentParser:
group_summary.add_argument(
"--metric", choices=("mean", "median", "sum", "count"), default="mean"
)
_add_output(group_summary, "analysis result", nested=True)

correlation = analysis_recipes.add_parser(
"correlation", help="compute complete-case Pearson correlation"
)
correlation.add_argument("--x-column", required=True)
correlation.add_argument("--y-column", required=True)
_add_output(correlation, "analysis result", nested=True)

meta_analysis = analysis_recipes.add_parser(
"meta-analysis", help="run DerSimonian-Laird random-effects pooling"
)
meta_analysis.add_argument("--effect-column", required=True)
meta_analysis.add_argument("--se-column", required=True)
meta_analysis.add_argument("--label-column")
_add_output(meta_analysis, "analysis result", nested=True)
data_analyze.set_defaults(handler=_command_data_analyze)
return parser

Expand Down
Loading