API Log Insights: Formats, Limits, Live Updates

Background. A cloud provider's API traffic is logged by many sources, each writing files in its own shape into input/<source>/: an edge server writes JSON Lines with api_key and latency_ms, a microservice writes CSV with client_id and response_time_ms. The service you are building reads those files, maps every entry onto one record (records.py), computes usage insights per source, and writes them to output/<source>/insights.json. sources.json says which format each source uses and how its field names map onto the record's (config.py reads and validates it). test_helpers.py builds a scratch workspace with both.

Implement parse in parsers.py and Engine in engine.py (metrics.py is yours to organise).

• parse(text, config, header=None) returns (records, skipped, header). jsonl: one JSON object per line, either a single entry or {"records": [...]} holding several; blank lines are nothing. csv: a header row names the columns, then one entry per row; pass the returned header back in when text continues a file whose header was already read, and the rows are then rows; for jsonl the returned header is None, for csv it is the list of column names. A records value that is not a list is one bad entry. An entry becomes a Record through the mapping; latency_ms and status must be integers, negative ones included (a digit string with an optional leading minus counts; floats, booleans, spaces and + do not); timestamp, client, endpoint and method must be strings (a null or a number is a missing field); the timestamp an ISO 8601 date-time with a UTC offset, kept as written. Anything that cannot become a record is one skipped: an unreadable line, a batch item, a CSV row with more or fewer cells than the header, a missing mapped field.
• Engine(input_dir, output_dir, sources_path, limits_path=None); run_once() reads every file of every configured source whose folder exists, and returns {source: insights} after writing each to output/<source>/insights.json. A folder with no configuration is ignored; a configured source with no folder is skipped; a bad configuration raises ValueError.
• Insights: {"source", "records", "skipped", "clients", "top_endpoints"}. clients maps each client id, in sorted order, to {"requests": n, "avg_latency_ms": sum / n} (a float). top_endpoints lists at most five {"endpoint", "requests"}, most requests first, equal counts by endpoint name. Clients and endpoints are per source.

Example

ws = Workspace()                                      # input/, output/, sources.json
ws.write_jsonl("EdgeServer", "log1.json", [edge_entry(), edge_entry(key="xyz789", latency=250)])
Engine(ws.input, ws.output, ws.sources_path).run_once()["EdgeServer"]
# {"source": "EdgeServer", "records": 2, "skipped": 0,
#  "clients": {"abc123": {"requests": 1, "avg_latency_ms": 120.0}, "xyz789": {"requests": 1, "avg_latency_ms": 250.0}},
#  "top_endpoints": [{"endpoint": "/v1/users", "requests": 2}]}

Done when Submit goes green.