d3-log-p95--x-ai-grok-4-7

Write a Python CLI `p95paths`: reads an access-log CSV with columns timestamp,path,latency_ms and prints the top-N paths by 95th-percentile latency (default N=10), highest first, as `path p95_ms count`. Use argparse for --top and the input file (stdin when omitted). Malformed rows are skipped with a warning count on stderr; an empty or header-only input prints nothing and exits 0. Deliver a complete, runnable script plus focused tests covering the percentile math and malformed-row handling. Accompany the code with a brief explanation of your design choices.

Prompt

Write a Python CLI `p95paths`: reads an access-log CSV with columns timestamp,path,latency_ms and prints the top-N paths by 95th-percentile latency (default N=10), highest first, as `path p95_ms count`. Use argparse for --top and the input file (stdin when omitted). Malformed rows are skipped with a warning count on stderr; an empty or header-only input prints nothing and exits 0. Deliver a complete, runnable script plus focused tests covering the percentile math and malformed-row handling. Accompany the code with a brief explanation of your design choices. ## Output contract First, write a brief explanation of your design choices — plain prose, no code fences. Then emit exactly one fenced code block per file, and no other text between or after the blocks: - The FIRST line of every block is that file's marker line: - TypeScript or JavaScript: `// FILE: <path>` - Python: `# FILE: <path>` - A single-file web app is one file named `index.html`, so its marker line is `// FILE: index.html`. - Everything after the marker line is that file complete, from its first line to its last — no diffs, no omissions, no placeholders. - Tag each fence with the file's language, for example ```html, ```typescript, ```python.

Artifact

Form: tree

p95paths.py

#!/usr/bin/env python3
"""Top-N request paths by 95th-percentile latency.

CSV columns: timestamp,path,latency_ms. File argument optional (stdin if
omitted). Prints "path  p95_ms  count", highest p95 first.
"""

import argparse
import csv
import math
import sys
from collections import defaultdict


def non_negative_int(value):
    try:
        number = int(value)
    except (TypeError, ValueError):
        raise argparse.ArgumentTypeError("must be an integer") from None
    if number < 0:
        raise argparse.ArgumentTypeError("must be >= 0")
    return number


def build_parser():
    parser = argparse.ArgumentParser(
        prog="p95paths",
        description="Print the top paths by 95th-percentile latency.",
    )
    parser.add_argument(
        "input",
        nargs="?",
        help="access-log CSV (timestamp,path,latency_ms); stdin if omitted or -",
    )
    parser.add_argument(
        "--top",
        type=non_negative_int,
        default=10,
        help="how many paths to print (default: 10)",
    )
    return parser


def percentile(values, pct):
    """Hyndman-Fan type-7 percentile (NumPy default linear interpolation).

    Position is (n - 1) * pct/100. The result blends the samples on either
    side of that position. pct is a percent in [0, 100].
    """
    if not values:
        raise ValueError("percentile of empty sample")
    if pct < 0 or pct > 100:
        raise ValueError("pct out of range")
    ordered = sorted(values)
    count = len(ordered)
    if count == 1:
        return float(ordered[0])
    rank = (count - 1) * (pct / 100.0)
    low = math.floor(rank)
    high = math.ceil(rank)
    if low == high:
        return float(ordered[low])
    weight = rank - low
    return float(ordered[low] * (1.0 - weight) + ordered[high] * weight)


def format_ms(value):
    rounded = round(float(value), 6)
    if rounded == int(rounded):
        return str(int(rounded))
    return f"{rounded:.6f}".rstrip("0").rstrip(".")


def _clean_cell(cell, index, first_row):
    text = cell
    if first_row and index == 0:
        text = text.lstrip("\ufeff")
    return text.strip()


def _is_blank(row):
    return not row or all(not cell.strip() for cell in row)


def _parse_latency(text):
    try:
        value = float(text)
    except (TypeError, ValueError):
        return None
    if not math.isfinite(value) or value < 0:
        return None
    return value


def read_log(fh):
    """Return (path -> [latency, ...], skipped_count)."""
    groups = defaultdict(list)
    skipped = 0
    header_checked = False
    for raw in csv.reader(fh):
        if _is_blank(raw):
            continue
        row = [_clean_cell(cell, i, not header_checked) for i, cell in enumerate(raw)]
        if not header_checked:
            header_checked = True
            if [cell.lower() for cell in row] == ["timestamp", "path", "latency_ms"]:
                continue
        if len(row) != 3:
            skipped += 1
            continue
        timestamp, path, latency_text = row
        latency = _parse_latency(latency_text)
        if timestamp == "" or path == "" or latency is None:
            skipped += 1
            continue
        groups[path].append(latency)
    return groups, skipped


def rank_paths(groups, top):
    ranked = []
    for path, latencies in groups.items():
        ranked.append((percentile(latencies, 95), len(latencies), path))
    ranked.sort(key=lambda item: (-item[0], -item[1], item[2]))
    if top == 0:
        return []
    return ranked[:top]


def main(argv=None):
    if argv is None:
        argv = sys.argv[1:]
    args = build_parser().parse_args(argv)
    try:
        if args.input and args.input != "-":
            with open(args.input, newline="", encoding="utf-8") as handle:
                groups, skipped = read_log(handle)
        else:
            groups, skipped = read_log(sys.stdin)
    except OSError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1
    if skipped:
        print(f"warning: skipped {skipped} malformed row(s)", file=sys.stderr)
    for p95, count, path in rank_paths(groups, args.top):
        print(f"{path}  {format_ms(p95)}  {count}")
    return 0


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

test_p95paths.py

#!/usr/bin/env python3
"""Focused tests for p95 math and malformed-row handling."""

import contextlib
import io
import sys
import tempfile
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
import p95paths


class PercentileTests(unittest.TestCase):
    def test_single_sample_is_itself(self):
        self.assertEqual(p95paths.percentile([7], 95), 7.0)

    def test_linear_interpolation_five_samples(self):
        # rank = (5 - 1) * 0.95 = 3.8 -> 4 * 0.2 + 5 * 0.8 = 4.8
        self.assertAlmostEqual(p95paths.percentile([1, 2, 3, 4, 5], 95), 4.8)

    def test_linear_interpolation_two_samples(self):
        # rank = 0.95 -> 10 * 0.05 + 20 * 0.95 = 19.5
        self.assertAlmostEqual(p95paths.percentile([20, 10], 95), 19.5)

    def test_linear_interpolation_with_plateau(self):
        # rank = 3 * 0.95 = 2.85 -> 1 * 0.15 + 100 * 0.85 = 85.15
        self.assertAlmostEqual(p95paths.percentile([100, 1, 1, 1], 95), 85.15)

    def test_unsorted_input_matches_sorted(self):
        values = [9, 1, 8, 2, 7, 3, 6, 4, 5]
        self.assertAlmostEqual(
            p95paths.percentile(values, 95),
            p95paths.percentile(sorted(values), 95),
        )

    def test_constant_sample(self):
        self.assertEqual(p95paths.percentile([4, 4, 4, 4], 95), 4.0)

    def test_format_ms_trims_integral_and_fractional(self):
        self.assertEqual(p95paths.format_ms(10.0), "10")
        self.assertEqual(p95paths.format_ms(p95paths.percentile([1, 2, 3, 4, 5], 95)), "4.8")
        self.assertEqual(p95paths.format_ms(p95paths.percentile([10, 20], 95)), "19.5")

    def test_empty_sample_rejected(self):
        with self.assertRaises(ValueError):
            p95paths.percentile([], 95)


class CliTests(unittest.TestCase):
    def _run(self, text, args):
        stdout, stderr = io.StringIO(), io.StringIO()
        with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr), contextlib.redirect_stdin(
            io.StringIO(text)
        ):
            code = p95paths.main(args)
        return code, stdout.getvalue(), stderr.getvalue()

    def _run_file(self, text, args):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "access.csv"
            path.write_text(text, encoding="utf-8")
            return self._run("", args + [str(path)])

    def test_empty_and_header_only_exit_quietly(self):
        for text in ("", "\n\n", "timestamp,path,latency_ms\n", "Timestamp,Path,Latency_ms\n"):
            code, out, err = self._run(text, [])
            self.assertEqual(code, 0, text)
            self.assertEqual(out, "")
            self.assertEqual(err, "")

    def test_top_zero_prints_nothing(self):
        text = "timestamp,path,latency_ms\n2020-01-01T00:00:00Z,/a,10\n"
        code, out, err = self._run(text, ["--top", "0"])
        self.assertEqual(code, 0)
        self.assertEqual(out, "")
        self.assertEqual(err, "")

    def test_malformed_rows_skipped_with_count(self):
        text = "\n".join(
            [
                "timestamp,path,latency_ms",
                "2020-01-01T00:00:00Z,/ok,10",
                "bad",
                "2020-01-01T00:00:01Z,/ok,notanumber",
                "2020-01-01T00:00:02Z,/ok,30",
                ",/missing-ts,10",
                "2020-01-01T00:00:03Z,,10",
                "2020-01-01T00:00:04Z,/ok,-1",
                "2020-01-01T00:00:05Z,/ok,20",
                "2020-01-01T00:00:06Z,/ok,inf",
                "2020-01-01T00:00:07Z,/extra,1,2",
                "",
            ]
        )
        code, out, err = self._run(text, ["--top", "10"])
        self.assertEqual(code, 0)
        # latencies 10, 30, 20 -> rank 1.9 -> 29
        self.assertEqual(out, "/ok  29  3\n")
        self.assertEqual(err, "warning: skipped 7 malformed row(s)\n")

    def test_file_input_and_tie_break(self):
        text = "\n".join(
            [
                "timestamp,path,latency_ms",
                "t,/b,10",
                "t,/b,20",
                "t,/a,10",
                "t,/a,20",
                "t,/a,10",
                "t,/a,20",
                "t,/c,10",
                "t,/c,20",
                "t,/low,1",
            ]
        )
        code, out, err = self._run_file(text, ["--top", "3"])
        self.assertEqual(code, 0)
        self.assertEqual(err, "")
        # same p95 19.5: higher count first, then path name; /low excluded by --top
        self.assertEqual(out, "/a  19.5  4\n/b  19.5  2\n/c  19.5  2\n")

    def test_default_top_is_ten_highest_first(self):
        lines = ["timestamp,path,latency_ms"]
        for i in range(12):
            lines.append(f"t,/p{i:02d},{i + 1}")
        code, out, err = self._run("\n".join(lines) + "\n", [])
        self.assertEqual(code, 0)
        self.assertEqual(err, "")
        printed = [line.split("  ") for line in out.splitlines()]
        self.assertEqual(len(printed), 10)
        self.assertEqual(printed[0], ["/p11", "12", "1"])
        self.assertEqual(printed[-1], ["/p02", "3", "1"])
        p95s = [int(cols[1]) for cols in printed]
        self.assertEqual(p95s, sorted(p95s, reverse=True))

    def test_quoted_path_and_bom_header(self):
        text = "\ufefftimestamp,path,latency_ms\n2020-01-01T00:00:00Z,\"/a b\",15\n"
        code, out, err = self._run(text, [])
        self.assertEqual(code, 0)
        self.assertEqual(err, "")
        self.assertEqual(out, "/a b  15  1\n")


if __name__ == "__main__":
    unittest.main()

Explanation

The tool is a single stdlib script: `argparse` for `--top` (default 10, reject negatives) and an optional CSV path, with stdin when the path is omitted or `-`. Rows are grouped by path; timestamp is required to be non-empty but is not ranked on. The 95th percentile is Hyndman–Fan type 7 (linear interpolation at `(n−1)×0.95`, same as NumPy’s default), so small samples stay continuous instead of jumping to a nearest rank. Ties break by higher count, then path name, so output is stable. Integral results print without a decimal; other values are rounded to six decimal places. Blank lines are ignored. A header row is recognized case-insensitively (BOM stripped) and is not a warning. Any other row that does not have exactly three fields, a non-empty path, and a finite latency ≥ 0 is skipped. One stderr line reports the skip count; dirty input still exits 0 so the report stays pipeable. Empty or header-only input prints nothing and exits 0. Tests call the percentile helper directly and drive the CLI for ordering, skips, and empty input.

Judge detail

No raw judge output is published for this item yet. When it is, it lands under results/raw/ and appears here verbatim; the rubric and protocol are already documented on the Methodology page.