d3-log-p95--openai-gpt-6-astra

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
"""Report the paths with the highest nearest-rank p95 latency."""

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


def p95(values):
    """Return the nearest-rank 95th percentile of a nonempty collection."""
    ordered = sorted(values)
    if not ordered:
        raise ValueError("p95 requires at least one value")
    rank = (95 * len(ordered) + 99) // 100
    return ordered[rank - 1]


def read_latencies(stream):
    """Return (latencies_by_path, malformed_row_count).

    The header must name timestamp, path, and latency_ms exactly once.
    Latencies must be finite and nonnegative; timestamps and paths must
    contain at least one non-whitespace character.
    """
    reader = csv.reader(stream, strict=True)
    groups = defaultdict(list)
    skipped = 0

    header = next(reader, None)
    if header is None:
        return groups, skipped
    if len(header) != 3 or set(header) != {"timestamp", "path", "latency_ms"}:
        raise ValueError(
            "CSV header must contain exactly timestamp,path,latency_ms"
        )

    timestamp_index = header.index("timestamp")
    path_index = header.index("path")
    latency_index = header.index("latency_ms")

    while True:
        try:
            row = next(reader)
        except StopIteration:
            break
        except csv.Error:
            # The reader has consumed the offending record. An unclosed
            # quoted field may consume multiple physical lines.
            skipped += 1
            continue

        if len(row) != len(header):
            skipped += 1
            continue

        timestamp = row[timestamp_index]
        path = row[path_index]
        if not timestamp.strip() or not path.strip():
            skipped += 1
            continue

        try:
            latency = float(row[latency_index])
        except ValueError:
            skipped += 1
            continue

        if not math.isfinite(latency) or latency < 0:
            skipped += 1
            continue

        groups[path].append(latency)

    return groups, skipped


def positive_int(text):
    try:
        value = int(text)
    except ValueError:
        raise argparse.ArgumentTypeError("must be a positive integer") from None
    if value <= 0:
        raise argparse.ArgumentTypeError("must be a positive integer")
    return value


def main(argv=None):
    parser = argparse.ArgumentParser(
        prog="p95paths",
        description="Print top paths by nearest-rank 95th-percentile latency.",
    )
    parser.add_argument(
        "--top",
        type=positive_int,
        default=10,
        metavar="N",
        help="number of paths to print (default: 10)",
    )
    parser.add_argument(
        "input",
        nargs="?",
        metavar="FILE",
        help="access-log CSV (default: stdin)",
    )
    args = parser.parse_args(argv)

    try:
        source = (
            open(args.input, encoding="utf-8", newline="")
            if args.input is not None
            else nullcontext(sys.stdin)
        )
        with source as stream:
            groups, skipped = read_latencies(stream)
    except (OSError, UnicodeError, ValueError, csv.Error) as exc:
        parser.error(str(exc))

    if skipped:
        noun = "row" if skipped == 1 else "rows"
        print(f"warning: skipped {skipped} malformed {noun}", file=sys.stderr)

    results = [
        (path, p95(latencies), len(latencies))
        for path, latencies in groups.items()
    ]
    results.sort(key=lambda item: (-item[1], item[0]))

    for path, percentile, count in results[:args.top]:
        print(f"{path}  {percentile:.2f}  {count}")

    return 0


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

test_p95paths.py

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

from p95paths import p95, read_latencies


SCRIPT = Path(__file__).with_name("p95paths.py")
HEADER = "timestamp,path,latency_ms\n"


def run_cli(text="", *args):
    return subprocess.run(
        [sys.executable, str(SCRIPT), *args],
        input=text,
        text=True,
        capture_output=True,
        check=False,
    )


class PercentileTests(unittest.TestCase):
    def test_single_value(self):
        self.assertEqual(p95([12.5]), 12.5)

    def test_small_sample_uses_maximum(self):
        self.assertEqual(p95([3, 1, 2]), 3)

    def test_exact_rank_boundary(self):
        # ceil(0.95 * 20) = 19, not 20 and not an interpolated value.
        self.assertEqual(p95(range(1, 21)), 19)

    def test_rank_rounds_up(self):
        # ceil(0.95 * 21) = 20.
        self.assertEqual(p95(range(1, 22)), 20)

    def test_sorting_does_not_mutate_input(self):
        values = [100, 2, 7, 2]
        self.assertEqual(p95(values), 100)
        self.assertEqual(values, [100, 2, 7, 2])

    def test_empty_collection_is_rejected(self):
        with self.assertRaises(ValueError):
            p95([])


class RowHandlingTests(unittest.TestCase):
    def test_bad_rows_are_skipped_and_counted(self):
        text = (
            HEADER
            + "t,/ok,10\n"
            + "t,/missing\n"
            + "t,/extra,12,unexpected\n"
            + "t,/bad,not-a-number\n"
            + "t,/bad,nan\n"
            + "t,/bad,inf\n"
            + "t,/bad,-1\n"
            + " ,/bad,5\n"
            + "t, ,5\n"
            + "\n"
            + '"bad"x,/bad,5\n'
            + "t,/ok,20\n"
        )
        groups, skipped = read_latencies(io.StringIO(text))
        self.assertEqual(dict(groups), {"/ok": [10.0, 20.0]})
        self.assertEqual(skipped, 10)

    def test_quoted_paths_and_reordered_header(self):
        text = 'path,latency_ms,timestamp\n"/a,b",0,t\n'
        groups, skipped = read_latencies(io.StringIO(text))
        self.assertEqual(dict(groups), {"/a,b": [0.0]})
        self.assertEqual(skipped, 0)

    def test_unterminated_quote_is_counted(self):
        groups, skipped = read_latencies(
            io.StringIO(HEADER + 't,"/unfinished,5\n')
        )
        self.assertEqual(dict(groups), {})
        self.assertEqual(skipped, 1)

    def test_invalid_header_is_rejected(self):
        with self.assertRaises(ValueError):
            read_latencies(io.StringIO("timestamp,path,duration\n"))


class CliTests(unittest.TestCase):
    def test_empty_and_header_only_inputs(self):
        for text in ("", HEADER):
            with self.subTest(text=text):
                result = run_cli(text)
                self.assertEqual(result.returncode, 0)
                self.assertEqual(result.stdout, "")
                self.assertEqual(result.stderr, "")

    def test_warning_and_valid_output_are_separated(self):
        result = run_cli(
            HEADER + "t,/ok,10\n" + "t,/bad,nope\n" + "t,/bad,-4\n"
        )
        self.assertEqual(result.returncode, 0)
        self.assertEqual(result.stdout, "/ok  10.00  1\n")
        self.assertEqual(result.stderr, "warning: skipped 2 malformed rows\n")

    def test_all_malformed_rows_produce_no_results(self):
        result = run_cli(HEADER + "t,/bad,nan\n")
        self.assertEqual(result.returncode, 0)
        self.assertEqual(result.stdout, "")
        self.assertEqual(result.stderr, "warning: skipped 1 malformed row\n")

    def test_top_order_ties_and_counts(self):
        text = HEADER + "t,/z,50\nt,/b,100\nt,/a,100\nt,/a,1\n"
        result = run_cli(text, "--top", "2")
        self.assertEqual(result.returncode, 0)
        self.assertEqual(result.stdout, "/a  100.00  2\n/b  100.00  1\n")
        self.assertEqual(result.stderr, "")

    def test_default_top_is_ten(self):
        text = HEADER + "".join(f"t,/p{i},{i}\n" for i in range(12))
        result = run_cli(text)
        self.assertEqual(result.returncode, 0)
        lines = result.stdout.splitlines()
        self.assertEqual(len(lines), 10)
        self.assertEqual(lines[0], "/p11  11.00  1")
        self.assertEqual(lines[-1], "/p2  2.00  1")

    def test_file_input(self):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "access.csv"
            path.write_text(HEADER + "t,/file,8.25\n", encoding="utf-8")
            result = run_cli("", str(path))
        self.assertEqual(result.returncode, 0)
        self.assertEqual(result.stdout, "/file  8.25  1\n")
        self.assertEqual(result.stderr, "")

    def test_nonpositive_top_is_rejected(self):
        result = run_cli(HEADER, "--top", "0")
        self.assertEqual(result.returncode, 2)
        self.assertEqual(result.stdout, "")
        self.assertIn("positive integer", result.stderr)


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

Explanation

The script uses the nearest-rank percentile: after sorting a path’s latencies, p95 is the value at rank `ceil(0.95 × count)`. Results are ordered by descending p95, with path names breaking ties, and printed to two decimal places. The CSV must have a header containing exactly the three named columns, in any order. Rows with missing fields, empty timestamps or paths, invalid CSV syntax, or non-finite/negative latencies are skipped; one aggregate warning is written to stderr. Timestamps are treated as opaque strings because no timestamp format was specified. Latencies are retained in memory for exact percentile calculation. Run with `python p95paths.py [--top N] [input.csv]`; run the tests with `python -m unittest -v test_p95paths.py`.

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.