py-02-csv-groupby

# CSV Group Sum

Implement a file **`solution.py`** containing a function:

```python
def group_sum(csv_text: str, key_col: str, val_col: str) -> dict[str, float]:
    ...
```

Parse `csv_text` as CSV and return a dictionary mapping each distinct value in
column `key_col` to the **sum** of column `val_col` for all rows with that key.

Rules:

- The **first non-blank line** is the header row naming the columns.
- Columns are comma-separated. `key_col` and `val_col` name two of those columns.
- Values in `val_col` are numeric (int or float); sum them as floats.
- **Ignore blank lines** anywhere in the input (including trailing newlines).
- If the input has only a header (or is empty), return an empty dict `{}`.
- You may assume `key_col` and `val_col` exist in the header.

Example:

```python
csv_text = "name,amount\\nalice,10\\nbob,5\\nalice,2.5\\n"
group_sum(csv_text, "name", "amount") == {"alice": 12.5, "bob": 5.0}
```

from solution import group_sum


def test_basic_sum():
    csv_text = "name,amount\nalice,10\nbob,5\nalice,2.5\n"
    assert group_sum(csv_text, "name", "amount") == {"alice": 12.5, "bob": 5.0}


def test_returns_floats():
    csv_text = "k,v\na,1\na,2\n"
    out = group_sum(csv_text, "k", "v")
    assert out == {"a": 3.0}
    assert all(isinstance(v, float) for v in out.values())


def test_ignores_blank_lines():
    csv_text = "\n\nname,amount\nalice,1\n\nbob,2\n\n\n"
    assert group_sum(csv_text, "name", "amount") == {"alice": 1.0, "bob": 2.0}


def test_header_only():
    assert group_sum("name,amount\n", "name", "amount") == {}


def test_empty_input():
    assert group_sum("", "name", "amount") == {}
    assert group_sum("\n\n", "name", "amount") == {}


def test_column_order_independent():
    csv_text = "amount,name\n10,x\n5,y\n3,x\n"
    assert group_sum(csv_text, "name", "amount") == {"x": 13.0, "y": 5.0}


def test_three_columns_picks_right_ones():
    csv_text = "region,product,sales\neast,a,100\nwest,b,50\neast,c,25\n"
    assert group_sum(csv_text, "region", "sales") == {"east": 125.0, "west": 50.0}


def test_negative_and_float_values():
    csv_text = "g,n\nx,-1.5\nx,0.5\ny,4\n"
    assert group_sum(csv_text, "g", "n") == {"x": -1.0, "y": 4.0}

You are an expert programmer. Solve the task exactly as specified. Output your solution as fenced code blocks using the required file name(s) and the exact function/type signatures requested. Prefer correctness; do not include prose outside code unless asked.

To solve this problem, we need to parse a CSV text and compute the sum of values in a specified column for each unique key in another specified column. The result should be a dictionary mapping each key to the sum of its corresponding values.

### Approach
1. **Parse the CSV Text**: Split the input CSV text into lines.
2. **Identify the Header**: The first non-blank line is considered the header. This line contains the column names.
3. **Locate Columns**: Determine the indices of the specified key and value columns from the header.
4. **Process Data Rows**: For each subsequent non-blank line, extract the key and value, convert the value to a float, and accumulate the sum for each key.
5. **Handle Edge Cases**: Ensure to handle cases where the input is empty or only contains the header by returning an empty dictionary.

### Solution Code
```python
def group_sum(csv_text: str, key_col: str, val_col: str) -> dict[str, float]:
    lines = csv_text.split('\n')
    header_line = None
    for line in lines:
        stripped = line.strip()
        if stripped:
            header_line = stripped
            break
    if not header_line:
        return {}
    header_parts = [part.strip() for part in header_line.split(',')]
    try:
        key_index = header_parts.index(key_col)
        val_index = header_parts.index(val_col)
    except ValueError:
        return {}
    result = {}
    for line in lines:
        stripped = line.strip()
        if not stripped:
            continue
        if line == header_line:
            continue
        parts = [part.strip() for part in line.split(',')]
        if len(parts) <= max(key_index, val_index):
            continue
        key = parts[key_index]
        value = float(parts[val_index])
        if key in result:
            result[key] += value
        else:
            result[key] = value
    return result
```

### Explanation
1. **Parsing the CSV Text**: The input string is split into individual lines using `split('\n')`.
2. **Finding the Header**: The first non-blank line is identified as the header. If no such line exists, an empty dictionary is returned.
3. **Extracting Column Indices**: The header is split into parts, and the indices of the specified key and value columns are found.
4. **Processing Each Line**: Each line is processed to extract the key and value. If a line is blank or does not contain enough data, it is skipped. The value is converted to a float and accumulated in the result dictionary.
5. **Returning the Result**: The accumulated sums are returned as a dictionary where each key maps to its corresponding sum.

This approach ensures that the CSV is parsed correctly, handles edge cases, and efficiently computes the required sums.

Okay, I need to solve this problem where I have to write a Python function called group_sum. The function takes a CSV text, a key column, and a value column, and returns a dictionary mapping each key to the sum of the corresponding values.

Let me think about the steps I need to take.

First, I need to parse the CSV text. The first non-blank line is the header. So, I should split the text into lines, ignoring any blank lines. Then, find the first non-empty line as the header.

Once I have the header, I can get the indices of the key_col and val_col. Because the columns are named in the header, I can split the header by commas and find the positions of key_col and val_col.

Then, for each subsequent line (ignoring blank lines), I need to split them into their columns. For each line, extract the key and value. The value is numeric, so I need to convert it to a float.

I'll create a dictionary where each key is the key from the CSV, and the value is the sum of all corresponding val_col entries.

Wait, but how do I handle the lines? I should process each line, skipping any that are blank. For each non-blank line, after the header, I'll split into parts, get the key and value, and accumulate the sum.

Let me outline the steps:

1. Split the csv_text into lines. I can use split('\n') for this.

2. Iterate through each line, skipping blank lines. So, for each line in lines, if line.strip() is not empty, process it.

3. The first non-blank line is the header. So, I'll find the header by checking each line until I find one that's not empty. Once found, split it into columns.

4. Find the indices of key_col and val_col in the header. So, for example, if the header is "name,amount", and key_col is "name", then index is 0.

5. For each remaining line (after the header), if it's not blank, split into parts. Then, get the key and value from their respective indices. Convert the value to float and add to the dictionary.

6. The dictionary starts as empty. For each key, if it's already present, add the value to the existing sum. Otherwise, create a new entry with the value.

Now, let's think about possible edge cases.

What if the CSV is empty? Then return empty dict.

What if the CSV has only the header? Then return empty dict.

What if a line has fewer elements than the header? Hmm, but according to the problem statement, I think we can assume that each line has the necessary columns. Or maybe not? Well, the problem says to assume that key_col and val_col exist in the header, but for the data rows, perhaps they might have missing values. But in that case, perhaps we should skip those rows or treat the value as zero? Wait, the problem says that the values in val_col are numeric, so perhaps each row has a value.

But perhaps in the data, a row might have fewer elements than the header. So, when splitting, if the row doesn't have enough elements, maybe we should skip that row.

So, during processing each data row, after splitting, if the length is less than the maximum of the key_col and val_col indices, then skip.

Alternatively, perhaps the CSV is well-formed, so each row has all the necessary columns. But to be safe, perhaps in the code, when processing a row, if the row doesn't have enough elements, we skip it.

Another point: the CSV could have lines with leading or trailing whitespace, but when splitting, the commas are the separators. So, when splitting each line, I should split on commas, and then strip whitespace from each field.

Wait, for example, a line like " alice , 10 " would split into [" alice ", " 10 "]. So, I should probably strip each field.

So, for each line, after splitting by commas, I can create a list where each element is stripped of whitespace.

So, for a line, parts = [part.strip() for part in line.split(',')]

Then, the key is parts[key_index], and the value is parts[val_index], which I convert to float.

So, putting it all together.

Let me outline the code structure.

Function group_sum(csv_text: str, key_col: str, val_col: str) -> dict[str, float]:

- Split csv_text into lines = csv_text.split('\n')

- Iterate through lines to find the first non-blank line (header_line). So, for line in lines: if line.strip() != '': break.

- If no header_line found (all lines are blank), return empty dict.

- Split header_line into header_parts = [part.strip() for part in header_line.split(',')]

- Find key_index = header_parts.index(key_col)

- Find val_index = header_parts.index(val_col)

- Initialize the result dictionary as empty.

- For each line in lines after the header_line:

   - if line.strip() is empty, skip.

   - else:

      - split into parts = [part.strip() for part in line.split(',')]

      - if len(parts) <= max(key_index, val_index): skip (not enough data)

      - key = parts[key_index]

      - value = float(parts[val_index])

      - add to the result: result[key] += value, or initialize if not present.

Wait, but how to process the lines after the header. Because the lines could be in any order, but the header is the first non-blank line.

So, perhaps I should process all lines, but the first non-blank is the header, and the rest are data lines, but only if they are non-blank.

Wait, no. Because the lines could be in any order, but the header is the first non-blank line. So, for example, the CSV could have blank lines before the header.

So, the approach is:

- Iterate through all lines, find the first non-blank line as header.

- Then, for all other lines (including those before the header), if they are non-blank, process them as data rows.

Wait, no. Because the lines before the header are considered as data rows? Or are they considered as part of the header?

No, the first non-blank line is the header. So, any lines before that are considered as blank and skipped. So, for example, if the CSV starts with a blank line, then the header is the next non-blank line.

So, the processing is:

- Iterate through all lines, find the first line that is not blank. That's the header.

- Then, for all other lines (including those before the header), if they are not blank, process them as data rows.

Wait, no. Becaus
…[8603 chars elided]…
der_line.split(',')]
    try:
        key_index = header_parts.index(key_col)
        val_index = header_parts.index(val_col)
    except ValueError:
        # According to the problem statement, key_col and val_col exist, so this shouldn't happen.
        return {}
    result = {}
    for line in lines:
        stripped = line.strip()
        if not stripped:
            continue
        if line == header_line:
            continue
        parts = [part.strip() for part in line.split(',')]
        if len(parts) <= max(key_index, val_index):
            continue
        key = parts[key_index]
        value = float(parts[val_index])
        if key in result:
            result[key] += value
        else:
            result[key] = value
    return result

Wait, but in the code, the line is compared to header_line as 'line == header_line'. But in the lines list, the header_line is the first non-blank line. So, for example, if the CSV is "name,amount\nalice,10\n", the lines are ["name,amount", "alice,10", ""]. So, in the for loop, when line is "name,amount", it's equal to header_line, so it's skipped.

Yes.

Testing the example:

csv_text = "name,amount\\nalice,10\\nbob,5\\nalice,2.5\\n"

Wait, in the example, the csv_text is "name,amount\nalice,10\nbob,5\nalice,2.5\n"

So, the code should process the lines correctly.

Another test case: what if a data row has more columns than needed? For example, "alice,10,extra". The code will process the key and val correctly, ignoring the extra.

Yes.

Another test case: what if a data row has fewer columns than needed? For example, "alice". Then, len(parts) is 1, which is less than max(0,1)=1 → 1 <=1 → True → skip.

Yes.

Another test case: what if the key is an empty string? Well, the problem says that key_col and val_col exist, but the key could be any string, including empty. So, the code handles that.

Another point: the function returns a dictionary with string keys and float values. So, the code correctly converts the value to float.

I think this should handle all cases.