diff --git a/backend/services/citation_utils.py b/backend/services/citation_utils.py index 9f36b53a..2193713a 100644 --- a/backend/services/citation_utils.py +++ b/backend/services/citation_utils.py @@ -7,6 +7,11 @@ from __future__ import annotations +import re +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Iterable + PREVIEW_MAX_CHARS = 300 @@ -31,9 +36,7 @@ def build_sources(docs: list[str], metas: list[dict]) -> list[dict]: key = (meta.get("source", "unknown"), meta.get("chunk", 0)) if key not in seen: preview = ( - doc[:PREVIEW_MAX_CHARS] + "..." - if len(doc) > PREVIEW_MAX_CHARS - else doc + doc[:PREVIEW_MAX_CHARS] + "..." if len(doc) > PREVIEW_MAX_CHARS else doc ) seen[key] = { "source": meta.get("source", "unknown"), @@ -41,3 +44,243 @@ def build_sources(docs: list[str], metas: list[dict]) -> list[dict]: "preview": preview, } return list(seen.values()) + + +# --------------------------------------------------------------------------- +# Source ranking (issue #933) +# --------------------------------------------------------------------------- + + +# Default authority multipliers — pushed into the ranking computation if the +# caller does not override. Filename pattern -> (regex, weight in [0,1]). +# Tested in `backend/tests/test_citation_ranking.py`. +DEFAULT_AUTHORITY_RULES: tuple[tuple[re.Pattern, float], ...] = ( + (re.compile(r"^README\.md$", re.IGNORECASE), 0.55), + (re.compile(r"^CONTRIBUTING\.md$", re.IGNORECASE), 0.50), + (re.compile(r"^docs/.*\.md$", re.IGNORECASE), 0.70), + (re.compile(r"^docs/.*\.mdx$", re.IGNORECASE), 0.70), + (re.compile(r"^docs/.*\.txt$", re.IGNORECASE), 0.65), + (re.compile(r"^backend/app\.py$", re.IGNORECASE), 0.50), + (re.compile(r"^backend/services/.*\.py$", re.IGNORECASE), 0.55), + (re.compile(r"^backend/routes/.*\.py$", re.IGNORECASE), 0.45), + (re.compile(r"^backend/tests/.*\.py$", re.IGNORECASE), 0.30), + (re.compile(r"^frontend/src/.*", re.IGNORECASE), 0.40), + (re.compile(r"\b(official|spec|rfc)\b", re.IGNORECASE), 0.85), +) + + +RELEVANCE_WEIGHT_DEFAULT = 0.65 +RECENCY_WEIGHT_DEFAULT = 0.10 +AUTHORITY_WEIGHT_DEFAULT = 0.25 + + +@dataclass +class RankWeights: + """Weighting for the three ranking signals. + + All three must lie in [0.0, 1.0] and the sum does NOT need to equal + 1.0 — the writer is normalised to total-weight during ranking. This + lets callers set, e.g. weights=(0.7, 0.1, 0.2) without including + their own renormalisation step. + """ + + relevance: float = RELEVANCE_WEIGHT_DEFAULT + recency: float = RECENCY_WEIGHT_DEFAULT + authority: float = AUTHORITY_WEIGHT_DEFAULT + + def normalised(self) -> "RankWeights": + total = self.relevance + self.recency + self.authority + if total <= 0: + return RankWeights(0.0, 0.0, 0.0) + return RankWeights( + relevance=self.relevance / total, + recency=self.recency / total, + authority=self.authority / total, + ) + + def as_dict(self) -> dict[str, float]: + return { + "relevance": self.relevance, + "recency": self.recency, + "authority": self.authority, + } + + +@dataclass +class RankedSource: + """One ranked source entry — the original source dict + the score + components used to rank it (so callers can display / audit).""" + + source: dict + score: float + components: dict[str, float] = field(default_factory=dict) + + def to_dict(self) -> dict: + return { + "source": self.source, + "score": self.score, + "components": self.components, + } + + +def _parse_isoformat(value: Any) -> datetime | None: + """Parse an ISO-8601 timestamp (or datetime) and force-UTC. + + Accepts: ISO strings (`2026-07-22T10:00:00Z`), naive datetime objects + (interpreted as UTC), aware datetime objects (converted to UTC). + Returns None for missing/empty/invalid inputs. + """ + if value is None or value == "": + return None + if isinstance(value, datetime): + dt = value + else: + try: + dt = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + except (TypeError, ValueError): + return None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt.astimezone(timezone.utc) + + +def _relevance_score(distance: Any) -> float: + """Convert a cosine distance into a relevance score in [0, 1]. + + ChromaDB's cosine distance is in [0, 2] where 0 == perfect match and + 2 == perfect opposite. We treat relevance = 1 - distance / 2.0, + clamped to [0, 1]. + """ + try: + d = float(distance) + except (TypeError, ValueError): + return 0.0 + if d != d: # NaN + return 0.0 + return max(0.0, min(1.0, 1.0 - d / 2.0)) + + +def _recency_score(created_at: Any, now: datetime | None = None) -> float: + """Convert a (recency) timestamp into a recency score in [0, 1]. + + Linear decay over a 365-day window: a source dated `now` scores 1.0, + one dated >1 year before `now` scores 0.0. Beyond the window the + score is 0.0; future-dated sources (clock skew) are clamped at 1.0. + """ + if created_at is None or created_at == "": + return 0.0 + when = _parse_isoformat(created_at) + if when is None: + return 0.0 + if now is None: + now = datetime.now(tz=timezone.utc) + delta_days = (now - when).total_seconds() / 86400.0 + if delta_days < 0: + return 1.0 + return max(0.0, 1.0 - delta_days / 365.0) + + +def _authority_score( + source_name: str, + rules: Iterable[tuple[re.Pattern, float]] = DEFAULT_AUTHORITY_RULES, + base: float = 0.40, +) -> float: + """Compute an authority score in [0, 1] for a source filename. + + Walks `rules` in order and returns the weight of the FIRST rule whose + pattern matches; falls back to `base` if no rule matches. Matches + against the rule's compiled regex object. + """ + if not source_name: + return base + for pattern, weight in rules: + if pattern.search(source_name): + return weight + return base + + +def rank_sources( + sources: list[dict], + *, + relevance_distances: list[float] | None = None, + recency_timestamps: list[str | datetime | None] | None = None, + weights: RankWeights | None = None, + authority_rules: tuple[tuple[re.Pattern, float], ...] = DEFAULT_AUTHORITY_RULES, + authority_base: float = 0.40, + now: datetime | None = None, +) -> list[dict]: + """Return a deterministic ranking of `sources`. + + Each returned dict has the shape: + + { + "source": , + "score": , + "components": { + "relevance": , + "recency": , + "authority": + } + } + + Ties on `score` are broken first by source `source` (filename) + alphabetically, then by `(chunk)` ascending. This is deliberately + deterministic so the test suite can pin the order without clock- + dependent flakiness. + + If `relevance_distances` is omitted or shorter than `sources`, + relevance defaults to 0.5 (neutral). Same for `recency_timestamps`. + + Setting all weight components to 0 effectively reduces ranking to + pure tie-breaking (filename + chunk order). + """ + if weights is None: + weights = RankWeights() + weights = weights.normalised() + + if relevance_distances is None: + relevance_distances = [] + if recency_timestamps is None: + recency_timestamps = [] + + ranked: list[RankedSource] = [] + for idx, source in enumerate(sources): + source_name = str(source.get("source", "")) + + rel_dist = relevance_distances[idx] if idx < len(relevance_distances) else None + rel = _relevance_score(rel_dist if rel_dist is not None else 1.0) + + rec_ts = recency_timestamps[idx] if idx < len(recency_timestamps) else None + rec = _recency_score(rec_ts, now=now) + + auth = _authority_score(source_name, rules=authority_rules, base=authority_base) + + composite = ( + weights.relevance * rel + weights.recency * rec + weights.authority * auth + ) + + ranked.append( + RankedSource( + source=dict(source), + score=composite, + components={ + "relevance": rel, + "recency": rec, + "authority": auth, + }, + ) + ) + + # Sort by score descending; tie-break by source filename ascending, then + # chunk ascending. Use a stable sort with explicit keys so the output + # order under equal inputs is independent of Python's TIMSORT stability + # for equal-primary-key cases. + def sort_key(rs: RankedSource): + return ( + -rs.score, + str(rs.source.get("source", "")), + int(rs.source.get("chunk", 0)), + ) + + ranked.sort(key=sort_key) + return [rs.to_dict() for rs in ranked] diff --git a/backend/tests/test_citation_ranking.py b/backend/tests/test_citation_ranking.py new file mode 100644 index 00000000..ca1a0f15 --- /dev/null +++ b/backend/tests/test_citation_ranking.py @@ -0,0 +1,383 @@ +"""Tests for source ranking (issue #933). + +Covers: + +- `_parse_isoformat` — ISO string, naive datetime, aware datetime, + missing/empty/invalid → None. +- `_relevance_score` — valid float, NaN, non-numeric → 0.0, perfect + match (0) → 1.0, worst (2) → 0.0. +- `_recency_score` — 0 days → 1.0, 365 days → 0.0, >365 days → 0.0, + missing/empty/invalid → 0.0, future (clock-skew) → 1.0. +- `_authority_score` — matches `README.md`, `docs/foo.md`, + `frontend/src/App.jsx`, `backend/services/foo.py`, unspecified + source → base (0.40), official-like keyword match. +- `RankWeights.normalised()` — 0/0/0 → all 0.0, 1/1/1 → 1/3 each. +- `rank_sources`: + - empty list → empty output. + - single source — score composited, rank=0. + - two sources with different relevance → higher relevance first. + - tie-breaking: sources with equal score → filename THEN chunk + ascending serve as deterministic tiebreakers. + - weights at 0 → all scores 0 → tie-on-filename/chunk order. + - custom authority rules override a doc. + - `now` kwarg pinning recency to a fixed date (for CI determinism). +- CLI surface: `scripts/rank_sources.py` (subprocess). +- Purity: no ChromaDB / sentence-transformers side effects. +""" + +from __future__ import annotations + +import json +import subprocess +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from services.citation_utils import ( # noqa: E402 + DEFAULT_AUTHORITY_RULES, + RankWeights, + _authority_score, + _parse_isoformat, + _recency_score, + _relevance_score, + rank_sources, +) + +# --------------------------------------------------------------------------- +# Singleton helpers +# --------------------------------------------------------------------------- + + +class TestParseIsoformat: + def test_iso_utc_z_suffix(self): + dt = _parse_isoformat("2026-07-22T10:00:00Z") + assert dt is not None + assert dt.tzinfo == timezone.utc + assert dt.year == 2026 and dt.day == 22 + + def test_naive_datetime_is_treated_as_utc(self): + dt = _parse_isoformat(datetime(2026, 7, 22, 10, 0, 0)) + assert dt is not None + assert dt.tzinfo == timezone.utc + + def test_aware_datetime_is_converted_to_utc(self): + eastern = timezone(timedelta(hours=-5)) + aware = datetime(2026, 7, 22, 10, 0, 0, tzinfo=eastern) + dt = _parse_isoformat(aware) + assert dt is not None + assert dt.tzinfo == timezone.utc + assert dt.hour == 15 # UTC + + def test_missing_and_empty_returns_none(self): + assert _parse_isoformat(None) is None + assert _parse_isoformat("") is None + + def test_garbage_string_returns_none(self): + assert _parse_isoformat("not-a-date") is None + + +class TestRelevanceScore: + def test_perfect_match_is_1(self): + assert _relevance_score(0.0) == 1.0 + + def test_worst_match_is_0(self): + assert _relevance_score(2.0) == 0.0 + + def test_midpoint_is_0_5(self): + assert _relevance_score(1.0) == 0.5 + + def test_nan_returns_0(self): + assert _relevance_score(float("nan")) == 0.0 + + def test_non_numeric_returns_0(self): + assert _relevance_score("abc") == 0.0 + assert _relevance_score(None) == 0.0 + + +class TestRecencyScore: + @pytest.fixture + def now(self) -> datetime: + return datetime(2026, 7, 23, 12, 0, 0, tzinfo=timezone.utc) + + def test_today_is_1(self, now): + assert _recency_score(now, now=now) == 1.0 + + def test_365_days_is_0(self, now): + assert _recency_score(now - timedelta(days=365), now=now) == 0.0 + + def test_older_than_365_is_0(self, now): + assert _recency_score(now - timedelta(days=400), now=now) == 0.0 + + def test_future_clock_skew_is_1(self, now): + assert _recency_score(now + timedelta(days=10), now=now) == 1.0 + + def test_missing_is_0(self, now): + assert _recency_score(None, now=now) == 0.0 + assert _recency_score("", now=now) == 0.0 + + def test_180_days_is_roughly_0_5(self, now): + score = _recency_score(now - timedelta(days=180), now=now) + assert score == pytest.approx(0.5, abs=0.02) + + +class TestAuthorityScore: + def test_readme_matches_dedicated_rule(self): + assert _authority_score("README.md") == 0.55 + + def test_docs_subdir_matches_high_authority(self): + assert _authority_score("docs/csrf-protection.md") == 0.70 + + def test_frontend_src_matches(self): + assert _authority_score("frontend/src/components/SettingsPanel.jsx") == 0.40 + + def test_backend_service_matches(self): + assert _authority_score("backend/services/rag_service.py") == 0.55 + + def test_unknown_source_gets_base(self): + assert _authority_score("notes.txt") == 0.40 + + def test_official_keyword_match_fires(self): + assert _authority_score("RFC 9110 spec.md") == 0.85 + + def test_empty_source_gets_base(self): + assert _authority_score("") == 0.40 + + def test_custom_rules_override(self): + import re + + rules = ((re.compile(r"^custom\.md$"), 1.0),) + assert _authority_score("custom.md", rules=rules) == 1.0 + + +class TestRankWeights: + def test_defaults_sum_to_1(self): + w = RankWeights() + assert w.relevance == 0.65 + assert w.recency == 0.10 + assert w.authority == 0.25 + + def test_normalised_handles_zero_sum(self): + w = RankWeights(0, 0, 0).normalised() + assert w.relevance == 0.0 + assert w.recency == 0.0 + assert w.authority == 0.0 + + def test_normalised_scales_proportionally(self): + w = RankWeights(0.7, 0.1, 0.2).normalised() + # Already sums to 1, so no change. + assert w.relevance == 0.7 + assert w.recency == 0.1 + assert w.authority == 0.2 + + def test_as_dict_round_trip(self): + w = RankWeights(0.6, 0.2, 0.2) + d = w.as_dict() + assert d == {"relevance": 0.6, "recency": 0.2, "authority": 0.2} + + +# --------------------------------------------------------------------------- +# rank_sources integration +# --------------------------------------------------------------------------- + + +class TestRankSources: + @pytest.fixture + def sample_sources(self) -> list[dict]: + return [ + {"source": "docs/csrf-protection.md", "chunk": 0, "preview": "CSRF..."}, + {"source": "README.md", "chunk": 2, "preview": "README preview..."}, + {"source": "README.md", "chunk": 0, "preview": "Intro..."}, + {"source": "uncategorized.txt", "chunk": 0, "preview": "Note..."}, + ] + + def test_empty_list_returns_empty(self): + assert rank_sources([]) == [] + + def test_single_source_has_score(self): + result = rank_sources([{"source": "a.md", "chunk": 0}]) + assert len(result) == 1 + assert "score" in result[0] + assert "components" in result[0] + assert result[0]["source"]["source"] == "a.md" + + def test_higher_relevance_ranks_first(self, sample_sources): + # Give the CSRF source the best relevance (lowest distance = 0), + # README entries higher distance → lower score. + dists = [0.0, 1.5, 1.5, 1.5] + result = rank_sources(sample_sources, relevance_distances=dists) + top = result[0] + assert top["source"]["source"] == "docs/csrf-protection.md" + + def test_tie_broken_by_filename_then_chunk(self, sample_sources): + # All relevance equal → authority dominates → README.md (0.55) + # > docs (0.70) the docs/ source wins. Between the two README + # entries chunk 0 beats chunk 2. uncategorized.txt = base 0.40. + result = rank_sources(sample_sources) + # Expected order: docs/csrf-protection.md (0.70 authority), README + # chunk 0, README chunk 2, uncategorized.txt (0.40). + assert result[0]["source"]["source"] == "docs/csrf-protection.md" + assert result[1]["source"]["chunk"] == 0 # README + assert result[2]["source"]["chunk"] == 2 # README + assert result[3]["source"]["source"] == "uncategorized.txt" + + def test_all_zero_weights_ranks_by_filename_then_chunk(self, sample_sources): + weights = RankWeights(0, 0, 0) + result = rank_sources(sample_sources, weights=weights) + # All scores are 0 → sorted exclusively by source name then chunk. + sources = [r["source"]["source"] for r in result] + chunks = [r["source"]["chunk"] for r in result] + assert sources == sorted(sources) + # Verify that two same-name entries are in chunk-order. + readme_idx = [i for i, s in enumerate(sources) if s == "README.md"] + assert len(readme_idx) == 2 + assert chunks[readme_idx[0]] < chunks[readme_idx[1]] + + def test_recency_dominates_when_weight_high(self, sample_sources): + weights = RankWeights(relevance=0.1, recency=0.8, authority=0.1) + now = datetime(2026, 7, 23, tzinfo=timezone.utc) + today = now.isoformat() + old = (now - timedelta(days=400)).isoformat() + recency_list = [today, today, old, old] + result = rank_sources( + sample_sources, + recency_timestamps=recency_list, + weights=weights, + now=now, + ) + # Top two must be the recent ones. + recent_sources = {result[0]["source"]["source"], result[1]["source"]["source"]} + assert "docs/csrf-protection.md" in recent_sources + assert ( + "README.md" in result[0]["source"]["source"] + or "README.md" in result[1]["source"]["source"] + ) + + def test_now_kwarg_freeze_date_determinism(self, sample_sources): + now = datetime(2026, 1, 1, tzinfo=timezone.utc) + result_a = rank_sources( + sample_sources, + recency_timestamps=["2026-01-01T00:00:00Z"] * 4, + now=now, + ) + result_b = rank_sources( + sample_sources, + recency_timestamps=["2026-01-01T00:00:00Z"] * 4, + now=now, + ) + assert result_a == result_b + + def test_custom_authority_rules_override(self, sample_sources): + import re + + rules = ( + (re.compile(r"^uncategorized"), 1.0), + *DEFAULT_AUTHORITY_RULES, + ) + result = rank_sources(sample_sources, authority_rules=rules) + # uncategorized should now be top (authority 1.0). + assert result[0]["source"]["source"] == "uncategorized.txt" + + def test_missing_relevance_defaults_to_neutral(self, sample_sources): + dists = [None, None, None, None] + result = rank_sources(sample_sources, relevance_distances=dists) + components = result[0]["components"] + assert components["relevance"] == pytest.approx(0.5) + + def test_short_relevance_list_padded_with_neutral(self, sample_sources): + dists = [0.0] # only first source has a distance + result = rank_sources(sample_sources, relevance_distances=dists) + assert result[0]["components"]["relevance"] == pytest.approx(1.0) + assert result[1]["components"]["relevance"] == pytest.approx(0.5) + + +# --------------------------------------------------------------------------- +# CLI tooling: scripts/rank_sources.py +# --------------------------------------------------------------------------- + + +class TestRankSourcesCLI: + @pytest.fixture + def tool_path(self) -> Path | None: + candidate = Path(__file__).resolve().parents[2] / "scripts" / "rank_sources.py" + return candidate if candidate.is_file() else None + + def run_cli(self, args: list[str]) -> subprocess.CompletedProcess: + tool = Path(__file__).resolve().parents[2] / "scripts" / "rank_sources.py" + assert tool.is_file() + return subprocess.run( + [sys.executable, str(tool), *args], + capture_output=True, + text=True, + check=False, + ) + + def test_json_input_returns_ranked_list(self, tmp_path: Path): + tool = Path(__file__).resolve().parents[2] / "scripts" / "rank_sources.py" + if not tool.is_file(): + pytest.skip("`scripts/rank_sources.py` not present in this checkout") + + in_file = tmp_path / "sources.json" + in_file.write_text( + json.dumps( + [ + {"source": "docs/a.md", "chunk": 0}, + {"source": "README.md", "chunk": 1}, + ] + ), + encoding="utf-8", + ) + result = subprocess.run( + [sys.executable, str(tool), "--json", str(in_file)], + capture_output=True, + text=True, + check=False, + ) + assert result.returncode == 0 + ranked = json.loads(result.stdout) + assert len(ranked) == 2 + assert ranked[0]["source"]["source"] == "docs/a.md" + + def test_plain_mode_output(self, tmp_path: Path): + tool = Path(__file__).resolve().parents[2] / "scripts" / "rank_sources.py" + if not tool.is_file(): + pytest.skip("`scripts/rank_sources.py` not present in this checkout") + + in_file = tmp_path / "sources.json" + in_file.write_text( + json.dumps( + [ + {"source": "docs/a.md", "chunk": 0}, + ] + ), + encoding="utf-8", + ) + result = subprocess.run( + [sys.executable, str(tool), "--plain", str(in_file)], + capture_output=True, + text=True, + check=False, + ) + assert result.returncode == 0 + assert "rank" in result.stdout.lower() or "#1" in result.stdout + + def test_missing_input_file_returns_2(self, tmp_path: Path): + tool = Path(__file__).resolve().parents[2] / "scripts" / "rank_sources.py" + if not tool.is_file(): + pytest.skip("`scripts/rank_sources.py` not present in this checkout") + + missing = tmp_path / "nonexistent.json" + result = subprocess.run( + [sys.executable, str(tool), str(missing)], + capture_output=True, + text=True, + check=False, + ) + assert result.returncode == 2 + + +if __name__ == "__main__": + raise SystemExit(pytest.main([__file__, "-v"])) diff --git a/scripts/rank_sources.py b/scripts/rank_sources.py new file mode 100644 index 00000000..94f25191 --- /dev/null +++ b/scripts/rank_sources.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 +""" +Rank sources offline — tooling companion to issue #933. + +Reads a JSON array of source objects (filename + chunk + optional metadata) +from stdin or a file, applies the relevance/recency/authority ranking +algorithm defined in `backend/services/citation_utils.py`, and prints +the ranked list. + +Usage: + python scripts/rank_sources.py --json sources.json + python scripts/rank_sources.py --plain sources.json + python scripts/rank_sources.py sources.json < same as --json + +Each source dict may include optional extra fields that the ranking +algorithm picks up: + + { + "source": "docs/csrf-protection.md", + "chunk": 0, + "distance": 0.23, // relevance distance (optional) + "created_at": "2026-07-20T12:00:00Z" // recency timestamp (optional) + } + +The extra fields mirror what ChromaDB's `query()` response can carry +per result when `include=["distances"]` is set (issue #933 does NOT +require changes to `rag_service.retrieve_context` to pass distances, +but the ranking function accepts them if present). + +Exit codes: 0 = ok, 1 = empty input (warn only), 2 = I/O or parse error. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "backend")) + +from services.citation_utils import ( # noqa: E402 + DEFAULT_AUTHORITY_RULES, + RankWeights, + rank_sources, +) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Rank a JSON source list via citation_utils.rank_sources." + ) + parser.add_argument( + "input", + type=Path, + nargs="?", + help="JSON file containing a list of source objects. Reads stdin if omitted.", + ) + parser.add_argument( + "--json", + action="store_true", + default=True, + help="Emit JSON output (default).", + ) + parser.add_argument( + "--plain", + action="store_true", + help="Emit a human-readable report.", + ) + parser.add_argument( + "--relevance-weight", + type=float, + default=0.65, + help="Weight for relevance signal (0.0-1.0, default 0.65).", + ) + parser.add_argument( + "--recency-weight", + type=float, + default=0.10, + help="Weight for recency signal (0.0-1.0, default 0.10).", + ) + parser.add_argument( + "--authority-weight", + type=float, + default=0.25, + help="Weight for authority signal (0.0-1.0, default 0.25).", + ) + args = parser.parse_args(argv) + + weights = RankWeights( + relevance=args.relevance_weight, + recency=args.recency_weight, + authority=args.authority_weight, + ) + + try: + if args.input and str(args.input) != "-": + raw = args.input.read_text(encoding="utf-8", errors="replace") + else: + raw = sys.stdin.read() + except OSError as exc: + print(f"error: failed to read input: {exc}", file=sys.stderr) + return 2 + + try: + sources = json.loads(raw) + except json.JSONDecodeError as exc: + print(f"error: invalid JSON: {exc}", file=sys.stderr) + return 2 + + if not isinstance(sources, list): + print("error: input must be a JSON list of source objects", file=sys.stderr) + return 2 + + if not sources: + print("warning: empty source list", file=sys.stderr) + if not args.plain: + print("[]") + return 1 + + # Pull optional distance + created_at from the input dicts so the + # user doesn't need to pass them as separate CLI args. + distances: list[float | None] = [] + timestamps: list[str | None] = [] + for src in sources: + dist = src.get("distance") + distances.append( + float(dist) if isinstance(dist, (int, float)) and dist == dist else None + ) + ts = src.get("created_at") + timestamps.append(str(ts) if ts else None) + + ranked = rank_sources( + sources, + relevance_distances=distances + if any(d is not None for d in distances) + else None, + recency_timestamps=timestamps if any(t for t in timestamps) else None, + weights=weights, + authority_rules=DEFAULT_AUTHORITY_RULES, + ) + + if args.plain: + print( + f"Ranked {len(ranked)} source(s) " + f"(relevance_w={weights.relevance:.2f} " + f"recency_w={weights.recency:.2f} " + f"authority_w={weights.authority:.2f}):" + ) + print() + for idx, entry in enumerate(ranked, start=1): + src = entry["source"] + comp = entry["components"] + print( + f" {idx:3d}. [{entry['score']:.3f}] {src['source']} " + f"chunk={src['chunk']} " + f"(rel={comp['relevance']:.2f} rec={comp['recency']:.2f} " + f"auth={comp['authority']:.2f})" + ) + else: + print(json.dumps(ranked, indent=2)) + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())