import unittest
import user_code
import ast
import re   
import importlib
import csv
import polars as pl

class TestTask(unittest.TestCase):
    def setUp(self):
        # Initializing local mock dataset using a Polars DataFrame matching the solution schema
        self.df = pl.DataFrame({
            "Publishers": ["Valve", "Valve", "IndieCo", "Valve"],
            "Price": [9.99, 0.0, 4.99, 19.99],
            "Positive": [1000, 500, 100, 2000]
        })

    def test_aggregation_columns_exist(self):
        import user_code
        importlib.reload(user_code)
        func = getattr(user_code, "aggregate_publishers_stats", None)
        _dynamic_test(
            self,
            callable(func),
            "aggregate_publishers_stats function exists and is callable.",
            "aggregate_publishers_stats function is missing or not callable."
        )
        result = func(self.df)
        _dynamic_test(
            self,
            result is not None and "Publishers" in result.columns,
            "Returned DataFrame contains grouping column 'Publishers'.",
            "Returned DataFrame is missing grouping column 'Publishers'."
        )
        _dynamic_test(
            self,
            "Average Price" in result.columns,
            "Returned DataFrame contains 'Average Price' metric column.",
            "Returned DataFrame is missing 'Average Price' metric column."
        )
        _dynamic_test(
            self,
            "Total Positive Reviews" in result.columns,
            "Returned DataFrame contains 'Total Positive Reviews' metric column.",
            "Returned DataFrame is missing 'Total Positive Reviews' metric column."
        )

    def test_aggregation_values_correct(self):
        import user_code
        importlib.reload(user_code)
        func = getattr(user_code, "aggregate_publishers_stats", None)
        result = func(self.df)
        
        if result is not None and "Average Price" in result.columns and "Total Positive Reviews" in result.columns:
            # Extract Valve records for validation
            valve_data = result.filter(pl.col("Publishers") == "Valve")
            
            _dynamic_test(
                self,
                len(valve_data) == 1,
                "Grouping strategy correctly combined duplicate records.",
                f"Expected unique row for 'Valve', found {len(valve_data)} rows."
            )
            
            # Valve math: 
            # Average Price = (9.99 + 0.0 + 19.99) / 3 = 9.99333...
            # Total Positive Reviews = 1000 + 500 + 2000 = 3500
            actual_price = valve_data["Average Price"][0]
            actual_reviews = valve_data["Total Positive Reviews"][0]
            
            _dynamic_test(
                self,
                abs(actual_price - 9.993333) < 1e-4,
                "Mean aggregation engine evaluated correct 'Average Price'.",
                f"Incorrect 'Average Price' calculation. Expected near 9.9933, Got: {actual_price}"
            )
            _dynamic_test(
                self,
                actual_reviews == 3500,
                "Sum aggregation engine evaluated correct 'Total Positive Reviews'.",
                f"Incorrect 'Total Positive Reviews' calculation. Expected 3500, Got: {actual_reviews}"
            )
            
            # IndieCo records validation
            indie_data = result.filter(pl.col("Publishers") == "IndieCo")
            actual_indie_price = indie_data["Average Price"][0]
            actual_indie_reviews = indie_data["Total Positive Reviews"][0]
            
            _dynamic_test(
                self,
                actual_indie_price == 4.99 and actual_indie_reviews == 100,
                "Single-instance group records aggregation validated cleanly.",
                f"Aggregation failed on single items. Got Price: {actual_indie_price}, Reviews: {actual_indie_reviews}"
            )
        else:
            _dynamic_test(self, False, "", "Metrics execution evaluation halted due to missing target columns.")

def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\s{2,}", " ", text)
    text = re.sub(r"\s*([,:?])\s*", r"\1 ", text)
    return text.strip()

def change_var(code: str, var_name: str, value: str) -> str:
    tree = ast.parse(code)
    lines = code.splitlines()
    changed = False
    assign_nodes = [
        (i, node)
        for i, node in enumerate(tree.body)
        if isinstance(node, ast.Assign)
        and any(isinstance(target, ast.Name) and target.id == var_name for target in node.targets)
    ]

    if not assign_nodes:
        return code

    for i, node in reversed(assign_nodes):
        start_line = node.lineno - 1
        line = lines[start_line]
        indent = ' ' * (len(line) - len(line.lstrip()))
        lines[start_line] = f"{indent}{var_name} = {value}"
        next_line = len(lines)
        for next_node in tree.body[i+1:]:
            if hasattr(next_node, 'lineno'):
                next_line = next_node.lineno - 1
                break
        if next_line > start_line + 1:
            lines[start_line+1:next_line] = []
        changed = True

    return '\\n'.join(lines) if changed else code

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

test_main.py

Python開発者向けの実践的な中級コースで、Polarsを用いたデータ整形技術を、実際のSteamゲームデータセットを使って習得します。Polarsのカラム指向パラダイム、選択、集約、結合、リシェイプ、そして文字列・日付・欠損値処理の基本操作を学びます。

Polarsと従来の行ベースDataFrameの主な違いを探求し、カラム指向の操作、選択、およびSteamゲームデータセットを用いた条件付きロジックに焦点を当てます。

Steamゲームおよびスパイインサイトデータセットを使用したPolarsの並列グループ化、集計、結合、リシェイプ機能の確実な理解を習得します。

データクリーニングの基本であるテキスト正規化、日付解析、Polarsにおける欠損値の堅牢な処理について学びます。

チャレンジ：Group_byと集約

解答