数据
处理

import json

payload = {
    "source": {
        "path": "data/raw/study.csv",
        "encoding": "utf-8",
        "version": "2026-08-27",
    },
    "rules": {
        "topic": "strip whitespace",
        "score": "0 <= score <= 100",
        "group": "topic",
    },
    "output": {
        "input_rows": 12,
        "cleaned_rows": 9,
        "rejected_rows": 3,
        "files": ["clean.csv", "rejected.csv", "summary.json"],
    },
}

print(json.dumps(payload, ensure_ascii=False, indent=2))

record = {
    "source_id": "python-basics-01",
    "topic": "Python",
    "minutes": 35,
    "score": 82,
}

required = {"source_id", "topic", "minutes", "score"}
missing = required - record.keys()
if missing:
    raise ValueError(f"缺少字段: {sorted(missing)}")

if not isinstance(record["minutes"], (int, float)):
    raise TypeError("minutes 必须是数字")
if record["minutes"] < 0:
    raise ValueError("minutes 不能为负数")

print("契约检查通过", record)

from importlib import metadata
from pathlib import Path
import platform
import sys

print("python:", sys.version.split()[0])
print("executable:", sys.executable)
print("platform:", platform.platform())
print("workdir:", Path.cwd())

for package in ("numpy", "pandas", "matplotlib"):
    try:
        print(f"{package}: {metadata.version(package)}")
    except metadata.PackageNotFoundError:
        print(f"{package}: NOT INSTALLED")

import json
from pathlib import Path

data_dictionary = {
    "topic": {
        "type": "string",
        "required": True,
        "nullable": False,
        "meaning": "一次记录所属的主题",
    },
    "minutes": {
        "type": "number",
        "unit": "minute",
        "minimum": 0,
        "nullable": False,
    },
    "score": {
        "type": "number",
        "unit": "0-100",
        "minimum": 0,
        "maximum": 100,
        "nullable": True,
    },
}

path = Path("output/data-dictionary.json")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
    json.dumps(data_dictionary, ensure_ascii=False, indent=2),
    encoding="utf-8",
)
print("fields:", sorted(data_dictionary))
print("score range:", data_dictionary["score"]["minimum"], "to",
      data_dictionary["score"]["maximum"])

import pandas as pd

frame = pd.DataFrame([
    {"topic": "Python", "minutes": 35, "score": 82},
    {"topic": "数据", "minutes": 50, "score": 91},
    {"topic": "Python", "minutes": 20, "score": 76},
])

print("shape:", frame.shape)
print("columns:", frame.columns.tolist())
print("dtypes:")
print(frame.dtypes)
print("missing:")
print(frame.isna().sum().to_dict())
print("unique topics:", frame["topic"].nunique())
print(frame.head(2).to_string(index=False))

import json
from pathlib import Path

summary = {
    "input_rows": 3,
    "topics": ["Python", "数据"],
    "rule": "score_ratio = score / 100",
    "sort": "score_ratio descending",
}

output = Path("output/experiment.json")
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(
    json.dumps(summary, ensure_ascii=False, indent=2),
    encoding="utf-8",
)

loaded = json.loads(output.read_text(encoding="utf-8"))
assert loaded["input_rows"] == 3, "输入行数记录错误"
print("可回读:", loaded)

import numpy as np

small = np.array([120, 120], dtype=np.int8)
wide = small.astype(np.int16)
float_values = np.array([0.1, 0.2, 0.3])

print("small:", small, "dtype:", small.dtype)
print("wide sum:", int(wide.sum()), "dtype:", wide.dtype)
print("float sum:", float_values.sum())
print("close to 0.6:", np.isclose(float_values.sum(), 0.6))

assert wide.dtype == np.int16
assert np.isclose(float_values.sum(), 0.6)

import numpy as np

values = np.array([[1, 2.5, 3], [4, 5, 6]])

print("ndim:", values.ndim)
print("shape:", values.shape)
print("size:", values.size)
print("dtype:", values.dtype)
print("itemsize:", values.itemsize)
print("nbytes:", values.nbytes)

assert values.size == values.shape[0] * values.shape[1]
assert values.ndim == 2

import numpy as np

indexes = np.arange(0, 5)
axis = np.linspace(0.0, 1.0, num=5)
buffer = np.full((2, 3), fill_value=-1, dtype=np.int16)

raw = np.array([1.9, 2.1, 3.8])
rounded = np.rint(raw).astype(np.int64)
truncated = raw.astype(np.int64)

print("indexes:", indexes)
print("axis:", axis)
print("buffer:", buffer)
print("rounded:", rounded)
print("truncated:", truncated)

import numpy as np

counts = np.arange(0, 10, 2)
axis = np.linspace(0, 1, num=5)
empty_scores = np.zeros((2, 3), dtype=np.float32)
rounded = np.array([1.9, 2.1]).astype(np.int64)

print(counts)
print(axis)
print(empty_scores.shape, empty_scores.dtype)
print(rounded)

import numpy as np

feature_names = ["minutes", "score", "completed"]
matrix = np.array([
    [35.0, 82.0, 1.0],
    [50.0, 91.0, 1.0],
])

assert matrix.ndim == 2
assert matrix.shape[1] == len(feature_names)

feature_mean = dict(zip(feature_names, matrix.mean(axis=0)))
print("feature order:", feature_names)
print("feature mean:", feature_mean)

reloaded_names = ["minutes", "score", "completed"]
assert reloaded_names == feature_names

import numpy as np

scores = np.array([60.0, 75.0, 90.0, np.nan])
valid = ~np.isnan(scores)

vectorized = np.where(valid, scores * 1.1, np.nan)

loop_result = []
for score in scores:
    if np.isnan(score):
        loop_result.append(np.nan)
    else:
        loop_result.append(score * 1.1)
loop_result = np.array(loop_result)

same = np.allclose(
    vectorized,
    loop_result,
    equal_nan=True,
)
print("same result:", same)
print("valid count:", int(valid.sum()))
print("scaled:", vectorized)

import numpy as np

matrix = np.array([
    [60.0, 70.0, np.nan],
    [90.0, 80.0, 75.0],
])

valid_count = np.sum(~np.isnan(matrix), axis=0)
mean = np.nanmean(matrix, axis=0)
minimum = np.nanmin(matrix, axis=0)
maximum = np.nanmax(matrix, axis=0)

for index, values in enumerate(zip(mean, valid_count, minimum, maximum)):
    average, count, low, high = values
    print({
        "test": index,
        "average": float(average),
        "valid_count": int(count),
        "min": float(low),
        "max": float(high),
    })

import numpy as np

scores = np.array([
    [60.0, 70.0, 80.0],
    [90.0, np.nan, 75.0],
])

row_mean = np.nanmean(scores, axis=1)
column_mean = np.nanmean(scores, axis=0)
row_count = np.sum(~np.isnan(scores), axis=1)
centered = scores - np.nanmean(scores, axis=1, keepdims=True)

print("row mean:", row_mean)
print("column mean:", column_mean)
print("valid count:", row_count)
print("centered shape:", centered.shape)

import numpy as np

minutes = np.array([20, 40, 60])
completed = np.array([True, False, True])

effective = np.where(completed, minutes, 0)
long_sessions = minutes >= 45
weighted = minutes * np.where(completed, 1.0, 0.5)

print("effective:", effective)
print("long sessions:", long_sessions)
print("weighted:", weighted)

import numpy as np

ids = np.arange(12)
rng = np.random.default_rng(7)
shuffled = rng.permutation(ids)

train = shuffled[:8]
valid = shuffled[8:10]
test = shuffled[10:]

print("train:", train)
print("valid:", valid)
print("test:", test)

assert set(train).isdisjoint(valid)
assert set(train).isdisjoint(test)
assert set(valid).isdisjoint(test)
assert len(set(train) | set(valid) | set(test)) == len(ids)

import numpy as np

matrix = np.arange(1, 13).reshape(3, 4)
print("matrix:\\n", matrix)
print("second row:", matrix[1])
print("first column:", matrix[:, 0])
print("middle block:\\n", matrix[0:2, 1:3])
print("rows in custom order:\\n", matrix[[2, 0]])

window = matrix[0:2, 0:2]
window[0, 0] = 99
print("changed by view:\\n", matrix)

independent = matrix[0:2, 0:2].copy()
independent[0, 0] = -1
print("original after copy change:\\n", matrix)
print("independent block:\\n", independent)

import numpy as np

scores = np.array([[60, 72, 81], [55, 90, 68], [88, 64, 92]])
valid = (scores >= 60) & (scores <= 100)
row_ok = valid.all(axis=1)
row_has_low = (scores < 60).any(axis=1)

bonus_by_test = np.array([5, 0, 2])
adjusted = scores + bonus_by_test
clipped = np.clip(adjusted, 0, 100)

print("valid mask:\\n", valid)
print("all tests valid:", row_ok)
print("has low score:", row_has_low)
print("adjusted:\\n", adjusted)
print("clipped:\\n", clipped)

import numpy as np

features = np.array([[1.0, 10.0, 100.0], [2.0, 20.0, 200.0]])
mean = features.mean(axis=0)
scale = features.std(axis=0)
scale = np.where(scale == 0, 1.0, scale)

normalized = (features - mean) / scale
assert normalized.shape == features.shape
print("mean:", mean)
print("normalized:\\n", normalized)

import numpy as np

def require_samples_features(array, feature_count):
    if array.ndim != 2:
        raise ValueError(f"需要二维数组，实际是 {array.shape}")
    if array.shape[1] != feature_count:
        raise ValueError(
            f"特征数不符，期望 {feature_count}，实际 {array.shape[1]}"
        )
    return array

features = np.arange(20).reshape(5, 4)
checked = require_samples_features(features, feature_count=4)
print("accepted:", checked.shape)

import numpy as np

def describe(name, array):
    print({
        "name": name,
        "shape": array.shape,
        "ndim": array.ndim,
        "dtype": str(array.dtype),
        "sample": array.reshape(-1)[:6].tolist(),
    })

raw = np.arange(12).reshape(3, 4)
column_vector = raw[:, 0:1]
column_flat = raw[:, 0]
row_mean = raw.mean(axis=1, keepdims=True)
transposed = raw.T

for name, array in [
    ("raw", raw),
    ("column_vector", column_vector),
    ("column_flat", column_flat),
    ("row_mean", row_mean),
    ("transposed", transposed),
]:
    describe(name, array)

assert column_vector.shape == (3, 1)
assert column_flat.shape == (3,)
assert row_mean.shape == (3, 1)

import numpy as np

numbered = np.arange(1, 13).reshape(3, 4)
as_two_rows = numbered.reshape(2, 6)
as_columns = numbered.reshape(12, 1)

print("numbered:\\n", numbered)
print("two rows:\\n", as_two_rows)
print("columns:\\n", as_columns)
print("transpose:\\n", numbered.T)

assert numbered.ravel()[0] == 1
assert as_two_rows.ravel()[5] == 6
assert as_columns.shape == (12, 1)

import numpy as np

raw = np.arange(12)
samples = raw.reshape(3, 4)
features = samples.T
one_feature = raw.reshape(-1, 1)
batch = one_feature.reshape(3, 4)

print("raw:", raw.shape)
print("samples:", samples.shape)
print("features after transpose:", features.shape)
print("one feature:", one_feature.shape)
print("batch:", batch.shape)

assert samples.size == raw.size
assert samples.shape == (3, 4)
assert one_feature.shape[1] == 1

import pandas as pd

first = pd.Series([35, 50, 20], index=["python", "data", "model"])
second = pd.Series([2, 3, 4], index=["model", "python", "data"])

print("first:")
print(first)
print("dtype:", first.dtype)
print("index unique:", first.index.is_unique)

by_label = first + second
by_position = pd.Series(
    first.to_numpy() + second.to_numpy(),
    index=first.index,
)

print("aligned by label:")
print(by_label)
print("explicitly by position:")
print(by_position)

duplicate = pd.Series([1, 2], index=["x", "x"])
print("duplicate lookup:")
print(duplicate["x"])

import pandas as pd

raw = pd.DataFrame({
    "score": [82, 91, 76],
    "topic": ["Python", "数据", "Python"],
    "row_id": ["a", "b", "c"],
})
raw.index.name = "source_row"

delivery = (
    raw.rename(columns={"row_id": "source_id"})
    .loc[:, ["source_id", "topic", "score"]]
    .sort_values(["topic", "source_id"])
    .reset_index(drop=True)
    .convert_dtypes()
)

print("raw index:", raw.index)
print("delivery index:", delivery.index)
print("columns:", delivery.columns.tolist())
print(delivery)
assert delivery.columns.tolist() == ["source_id", "topic", "score"]
assert delivery["source_id"].is_unique

import pandas as pd

frame = pd.DataFrame([
    {"topic": " Python ", "minutes": "35", "score": 82},
    {"topic": "数据", "minutes": "50", "score": 91},
    {"topic": "Python", "minutes": "unknown", "score": None},
])

raw = frame.copy()
frame["topic"] = frame["topic"].astype("string").str.strip()
frame["minutes"] = pd.to_numeric(frame["minutes"], errors="coerce")
frame["score"] = pd.to_numeric(frame["score"], errors="coerce")

print("shape:", frame.shape)
print("dtypes:\\n", frame.dtypes)
print("missing:\\n", frame.isna().sum())
print("topics:", frame["topic"].value_counts(dropna=False).to_dict())
print("conversion failures:", frame["minutes"].isna().sum())

assert raw.shape == frame.shape
assert {"topic", "minutes", "score"}.issubset(frame.columns)

import pandas as pd

df = pd.DataFrame({
    "topic": ["Python", "数据", "模型", "Python", "数据"],
    "minutes": [35, 50, None, 20, 65],
    "score": [82, 91, 88, 76, None],
    "completed": [True, True, True, False, True],
})

score_ok = df["score"].ge(80) & df["score"].notna()
topic_ok = df["topic"].isin(["Python", "数据"])
mask = score_ok & topic_ok

selected = df.loc[
    mask,
    ["topic", "minutes", "score", "completed"],
].copy()
selected["score_per_minute"] = (
    selected["score"] / selected["minutes"]
)
selected = selected.loc[
    selected["minutes"].gt(0)
].sort_values(
    ["score", "topic"],
    ascending=[False, True],
).reset_index(drop=True)

print("selected shape:", selected.shape)
print(selected)
assert df["score"].isna().sum() == 1
assert selected["score_per_minute"].notna().all()

import pandas as pd

df = pd.DataFrame({
    "topic": ["Python", "数据", "模型", "Python"],
    "minutes": [35, 50, 15, 20],
    "score": [82, 91, 58, 76],
})

high = df.loc[df["score"] >= 80, ["topic", "score"]]
first_two = df.iloc[0:2, :]
mask = (df["minutes"] > 20) & df["topic"].isin(["Python", "数据"])

df.loc[mask, "level"] = "重点"
df.loc[~mask, "level"] = "普通"
subset = df.loc[df["score"] < 80].copy()
subset["score"] = subset["score"] + 5

print("high:\\n", high)
print("first two:\\n", first_two)
print("updated:\\n", df)
print("independent subset:\\n", subset)

import pandas as pd

df = pd.DataFrame({
    "topic": ["Python", "数据", "Python"],
    "score": [82, 91, 76],
})

high = df.loc[df["score"] >= 80, ["topic", "score"]]
df.loc[df["score"] < 80, "level"] = "需要复习"
df.loc[df["score"] >= 80, "level"] = "达标"

print(high)
print(df)

import json
from pathlib import Path
import pandas as pd

root = Path("data")
csv_path = root / "raw" / "study.csv"
json_path = root / "raw" / "study.json"

if not csv_path.exists():
    raise FileNotFoundError(f"找不到输入文件: {csv_path.resolve()}")

frame = pd.read_csv(
    csv_path,
    encoding="utf-8",
    na_values=["", "NA", "null"],
)
print("csv shape:", frame.shape)
print("csv columns:", frame.columns.tolist())

if json_path.exists():
    payload = json.loads(json_path.read_text(encoding="utf-8"))
    records = payload if isinstance(payload, list) else payload.get("records", [])
    json_frame = pd.DataFrame.from_records(records)
    print("json shape:", json_frame.shape)

import pandas as pd

raw = pd.DataFrame({
    "minutes_text": [" 35 ", "1,200", "unknown", None],
    "score_text": ["82", "91", "82%", "—"],
    "recorded_text": ["2026-08-01", "2026-08-02", "bad-date", None],
    "completed_text": ["yes", "no", "Y", "unknown"],
})

data = raw.copy()
data["minutes"] = pd.to_numeric(
    data["minutes_text"].str.replace(",", "", regex=False),
    errors="coerce",
)
data["score"] = pd.to_numeric(
    data["score_text"].str.rstrip("%"),
    errors="coerce",
)
data["recorded_at"] = pd.to_datetime(
    data["recorded_text"],
    format="%Y-%m-%d",
    errors="coerce",
)
completed_map = {"yes": True, "y": True, "no": False, "n": False}
data["completed"] = data["completed_text"].str.lower().map(completed_map)

for column in ["minutes", "score", "recorded_at", "completed"]:
    print(column, "failed:", int(data[column].isna().sum()))

print(data)

from pathlib import Path
import pandas as pd

path = Path("data/raw/study.csv")
required = {"topic", "minutes", "score"}
total_rows = 0
rejected_rows = 0
partial_summaries = []

for chunk_no, chunk in enumerate(
    pd.read_csv(path, chunksize=1000, encoding="utf-8"),
    start=1,
):
    total_rows += len(chunk)
    missing = required - set(chunk.columns)
    if missing:
        raise ValueError(f"第 {chunk_no} 块缺少列: {sorted(missing)}")

    chunk["minutes"] = pd.to_numeric(
        chunk["minutes"], errors="coerce"
    )
    chunk["score"] = pd.to_numeric(
        chunk["score"], errors="coerce"
    )
    invalid = (
        chunk["minutes"].isna()
        | chunk["minutes"].lt(0)
        | chunk["score"].isna()
        | chunk["score"].lt(0)
        | chunk["score"].gt(100)
    )
    rejected_rows += int(invalid.sum())
    valid = chunk.loc[~invalid]
    partial_summaries.append(
        valid.groupby("topic")
        .agg(records=("score", "size"), score_sum=("score", "sum"))
    )

print("input rows:", total_rows)
print("rejected rows:", rejected_rows)
print("chunks:", len(partial_summaries))

import pandas as pd

def validate_table(frame, required, unique_key=None):
    problems = []
    missing_columns = sorted(set(required) - set(frame.columns))
    if missing_columns:
        problems.append(f"缺少列: {missing_columns}")
        return frame.iloc[0:0].copy(), frame.copy(), problems

    data = frame.copy()
    data["minutes"] = pd.to_numeric(data["minutes"], errors="coerce")
    data["score"] = pd.to_numeric(data["score"], errors="coerce")

    invalid = (
        data["topic"].isna()
        | data["topic"].astype("string").str.strip().eq("")
        | data["minutes"].isna()
        | data["minutes"].lt(0)
        | data["score"].isna()
        | data["score"].lt(0)
        | data["score"].gt(100)
    )

    if unique_key is not None:
        duplicated = data[unique_key].duplicated(keep=False)
        invalid = invalid | duplicated

    clean = data.loc[~invalid].copy()
    rejected = data.loc[invalid].copy()
    return clean, rejected, problems

sample = pd.DataFrame({
    "topic": ["Python", "数据", ""],
    "minutes": [35, -2, 20],
    "score": [82, 91, 105],
})
cleaned, rejected, problems = validate_table(
    sample,
    required=["topic", "minutes", "score"],
)
print("cleaned:", len(cleaned))
print("rejected:", len(rejected))
print("problems:", problems)

def validate_table(frame, required_columns):
    missing = [
        column for column in required_columns
        if column not in frame.columns
    ]
    if missing:
        raise ValueError(f"缺少字段：{missing}")
    if frame.empty:
        raise ValueError("表格为空，不能继续分析")
    if frame["minutes"].isna().any():
        raise ValueError("minutes 含有无法转换的值")
    if (frame["minutes"] < 0).any():
        raise ValueError("minutes 不能为负数")
    return frame

required = ["topic", "minutes", "score"]
checked = validate_table(df, required)
print("validated rows:", len(checked))

import pandas as pd

frame = pd.DataFrame({
    "topic": ["Python", "数据", "模型", "Python", "数据"],
    "minutes": [35, None, 50, 20, 0],
    "score": [82, 91, None, None, 76],
    "completed": [True, True, False, None, True],
})

missing = frame.isna()
report = pd.DataFrame({
    "missing_count": missing.sum(),
    "missing_rate": missing.mean(),
})
print(report)

frame["score_missing"] = frame["score"].isna()
frame["minutes_missing"] = frame["minutes"].isna()

summary = (
    frame.groupby("topic", dropna=False)
    .agg(
        records=("topic", "size"),
        valid_scores=("score", "count"),
        average_score=("score", "mean"),
    )
    .reset_index()
)
print(summary)

import pandas as pd

events = pd.DataFrame([
    {"source_id": "a", "day": "2026-08-01", "minutes": 30, "updated_at": 1},
    {"source_id": "a", "day": "2026-08-01", "minutes": 30, "updated_at": 1},
    {"source_id": "b", "day": "2026-08-01", "minutes": 20, "updated_at": 1},
    {"source_id": "b", "day": "2026-08-01", "minutes": 35, "updated_at": 2},
])

all_columns = events.duplicated(keep=False)
key_columns = events.duplicated(
    subset=["source_id", "day"],
    keep=False,
)
print("exact duplicate rows:", int(all_columns.sum()))
print("duplicate keys:", int(key_columns.sum()))
print(events.loc[key_columns].sort_values(["source_id", "updated_at"]))

ordered = events.sort_values("updated_at")
latest = ordered.drop_duplicates(
    subset=["source_id", "day"],
    keep="last",
)
print("latest per key:")
print(latest)

import pandas as pd

df = pd.DataFrame({
    "source_id": ["a", "a", "b", "c"],
    "title": [" Python ", "Python", "模型", "模型"],
    "score": [80, 80, 92, 92],
})

df["title_key"] = df["title"].str.strip().str.casefold()
duplicates = df[df.duplicated(
    subset=["source_id", "title_key"],
    keep=False,
)]
print(duplicates.sort_values(["source_id", "title_key"]))

import pandas as pd

raw = pd.DataFrame({
    "row_id": ["a", "b", "c", "d"],
    "topic": [" Python ", "数据", "", "Python"],
    "minutes_text": ["35", "-2", "20", "unknown"],
    "score_text": ["82", "91", "105", "76"],
})

data = raw.copy()
audit = pd.DataFrame({
    "row_id": data["row_id"],
    "original_topic": data["topic"],
    "original_minutes": data["minutes_text"],
    "original_score": data["score_text"],
})

data["topic"] = data["topic"].astype("string").str.strip()
data["minutes"] = pd.to_numeric(data["minutes_text"], errors="coerce")
data["score"] = pd.to_numeric(data["score_text"], errors="coerce")

reason = pd.Series("", index=data.index, dtype="string")
checks = [
    (data["topic"].eq(""), "empty_topic"),
    (data["minutes"].isna(), "minutes_not_numeric"),
    (data["minutes"].lt(0), "minutes_negative"),
    (data["score"].isna(), "score_not_numeric"),
    (data["score"].gt(100), "score_out_of_range"),
]
for mask, label in checks:
    reason = reason.mask(mask & reason.eq(""), label)

data["reject_reason"] = reason
audit["reject_reason"] = reason
clean = data.loc[reason.eq("")].copy()
rejected = audit.loc[reason.ne("")].copy()

print("clean:", len(clean))
print("rejected:")
print(rejected)
assert len(clean) + len(rejected) == len(raw)

import pandas as pd

data = pd.DataFrame({
    "topic": ["Python", "数据", "模型", "Python"],
    "minutes": [35, -4, 900, 20],
    "score": [82, 105, 88, "unknown"],
})

data["minutes_num"] = pd.to_numeric(data["minutes"], errors="coerce")
data["score_num"] = pd.to_numeric(data["score"], errors="coerce")

reasons = pd.Series("", index=data.index, dtype="string")
rules = [
    (data["minutes_num"].isna(), "minutes_not_numeric"),
    (data["minutes_num"].lt(0), "minutes_negative"),
    (data["minutes_num"].gt(24 * 60), "minutes_too_large"),
    (data["score_num"].isna(), "score_not_numeric"),
    (data["score_num"].lt(0) | data["score_num"].gt(100), "score_out_of_range"),
]

for mask, reason in rules:
    reasons = reasons.mask(mask & reasons.eq(""), reason)

data["reject_reason"] = reasons
rejected = data.loc[data["reject_reason"].ne("")]
clean = data.loc[data["reject_reason"].eq("")]
print("clean rows:", len(clean))
print("rejected rows:")
print(rejected[["topic", "reject_reason"]])

import pandas as pd

events = pd.DataFrame({
    "topic": ["Python", "Python", "数据", "数据", "数据"],
    "minutes": [35, 20, 50, None, 40],
    "score": [82, 76, 91, 88, None],
    "source_id": ["a", "a", "b", "c", "c"],
})

summary = (
    events.groupby("topic", dropna=False)
    .agg(
        records=("topic", "size"),
        valid_scores=("score", "count"),
        average_score=("score", "mean"),
        total_minutes=("minutes", "sum"),
        sources=("source_id", "nunique"),
    )
    .reset_index()
    .sort_values(["average_score", "records"], ascending=[False, False])
)

print(summary)
print("input rows:", len(events))
print("group records:", int(summary["records"].sum()))

import pandas as pd

events = pd.DataFrame({
    "source_id": ["a", "b", "c", "x"],
    "minutes": [35, 50, 20, 15],
})
topics = pd.DataFrame({
    "source_id": ["a", "b", "c"],
    "topic": ["Python", "数据", "模型"],
})

before_rows = len(events)
before_total = events["minutes"].sum()

joined = events.merge(
    topics,
    on="source_id",
    how="left",
    validate="many_to_one",
    indicator=True,
)

print(joined)
print("match status:", joined["_merge"].value_counts().to_dict())
print("unmatched:", int((joined["_merge"] == "left_only").sum()))
assert len(joined) == before_rows
assert joined["minutes"].sum() == before_total

import pandas as pd

documents = pd.DataFrame({
    "source_id": ["a", "b", "c"],
    "title": ["Python", "数据", "模型"],
})
relations = pd.DataFrame({
    "source_id": ["a", "a", "b", "c", "c"],
    "tag": ["code", "beginner", "table", "model", "beginner"],
})
tags = pd.DataFrame({
    "tag": ["code", "beginner", "table", "model"],
    "description": ["代码", "入门", "表格", "模型"],
})

assert not relations.duplicated(["source_id", "tag"]).any()
joined = (
    relations
    .merge(documents, on="source_id", how="left",
           validate="many_to_one")
    .merge(tags, on="tag", how="left",
           validate="many_to_one")
)
print(joined)
print("documents:", joined["source_id"].nunique())
print("relations:", len(joined))
assert joined["title"].notna().all()
assert joined["description"].notna().all()

import pandas as pd

events = pd.DataFrame({
    "topic": ["Python", "Python", "数据", "数据"],
    "week": [1, 2, 1, 2],
    "minutes": [35, 20, 50, 40],
})

print("raw shape:", events.shape)
print("raw key duplicates:",
      int(events.duplicated(["topic", "week"]).sum()))

long = events.copy()
long["metric"] = "minutes"
long = long.rename(columns={"minutes": "value"})
print("long shape:", long.shape)

wide = long.pivot(
    index="topic",
    columns="week",
    values="value",
)
wide.columns = [f"week_{number}" for number in wide.columns]
wide = wide.reset_index()
print("wide shape:", wide.shape)
print(wide)

assert set(wide["topic"]) == set(events["topic"])
assert wide.shape[0] == events["topic"].nunique()

import pandas as pd

wide = pd.DataFrame({
    "topic": ["Python", "数据"],
    "week_1": [35, 50],
    "week_2": [20, 40],
})

long = wide.melt(
    id_vars=["topic"],
    var_name="week",
    value_name="minutes",
)
long["week_no"] = long["week"].str.extract(r"(\\d+)").astype(int)

report = long.pivot_table(
    index="topic",
    columns="week",
    values="minutes",
    aggfunc="sum",
    fill_value=0,
).reset_index()

print("long:")
print(long.sort_values(["topic", "week_no"]))
print("report:")
print(report)

import pandas as pd

wide = pd.DataFrame({
    "topic": ["Python", "数据"],
    "week_1": [35, 50],
    "week_2": [20, 40],
})

long = wide.melt(
    id_vars=["topic"],
    var_name="week",
    value_name="minutes",
)
report = long.pivot_table(
    index="topic",
    columns="week",
    values="minutes",
    aggfunc="sum",
    fill_value=0,
)
print(long)
print(report)

import pandas as pd

summary = pd.DataFrame({
    "topic": ["Python", "数据", "模型"],
    "records": [12, 8, 5],
    "average_score": [82.5, 91.0, 88.0],
})

question = "比较每个主题有多少条有效学习记录"
if "多少" in question or "比较" in question:
    chart = "bar"
elif "变化" in question:
    chart = "line"
elif "分布" in question:
    chart = "hist"
else:
    chart = "scatter"

print("question:", question)
print("chart:", chart)
print("columns available:", summary.columns.tolist())

from pathlib import Path
import matplotlib.pyplot as plt
import pandas as pd

summary = pd.DataFrame({
    "topic": ["模型", "数据", "Python"],
    "average_score": [95.0, 89.5, 79.0],
    "records": [3, 8, 12],
})

output_dir = Path("output")
output_dir.mkdir(parents=True, exist_ok=True)

fig, ax = plt.subplots(figsize=(7, 4))
bars = ax.bar(summary["topic"], summary["average_score"], color="#176b66")
ax.set_title("各主题平均分（附样本量）")
ax.set_xlabel("主题")
ax.set_ylabel("平均分（0–100）")
ax.set_ylim(0, 100)
ax.grid(axis="y", alpha=0.25)

for bar, records in zip(bars, summary["records"]):
    ax.text(
        bar.get_x() + bar.get_width() / 2,
        bar.get_height() + 2,
        f"n={records}",
        ha="center",
        va="bottom",
        fontsize=9,
    )

fig.tight_layout()
path = output_dir / "average-score.png"
fig.savefig(path, dpi=160, facecolor="white")
plt.close(fig)
print("saved:", path, "bytes:", path.stat().st_size)

import json
from pathlib import Path
import pandas as pd

summary = pd.DataFrame({
    "topic": ["Python", "数据", "模型"],
    "records": [12, 8, 3],
    "average_score": [82.5, 91.0, 88.0],
})

output = Path("output")
output.mkdir(parents=True, exist_ok=True)
summary.to_csv(output / "summary.csv", index=False)
(output / "summary.json").write_text(
    json.dumps(summary.to_dict(orient="records"), ensure_ascii=False, indent=2),
    encoding="utf-8",
)
loaded = pd.read_json(output / "summary.json")
assert loaded["records"].sum() == summary["records"].sum()
print("bundle rows:", len(loaded))

import json
from pathlib import Path

report = {
    "observation": "数据主题有效记录更多",
    "evidence": {
        "input_rows": 20,
        "valid_rows": 17,
        "groups": 4,
        "missing_score_rows": 2,
    },
    "interpretation": "样本量和资料难度可能共同影响平均分",
    "limitation": "示例数据不是总体样本",
    "next_action": "补充按周记录并抽查拒绝行",
}

path = Path("output/report-snapshot.json")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
    json.dumps(report, ensure_ascii=False, indent=2),
    encoding="utf-8",
)
loaded = json.loads(path.read_text(encoding="utf-8"))
assert set(["observation", "evidence", "limitation"]).issubset(loaded)
print(loaded["observation"])

from pathlib import Path
import matplotlib.pyplot as plt
import pandas as pd

summary = pd.DataFrame({
    "topic": ["Python", "数据分析", "模型基础"],
    "average": [82.5, 91.0, 88.0],
    "count": [12, 8, 3],
    "minimum": [60, 78, 80],
    "maximum": [98, 100, 95],
})

fig, ax = plt.subplots(figsize=(8, 4.5))
bars = ax.barh(
    summary["topic"],
    summary["average"],
    color=["#176b66", "#a15f2d", "#5d5a8e"],
)
ax.set_title("学习主题平均分（有效记录数见柱末）")
ax.set_xlabel("平均分（0–100）")
ax.set_xlim(0, 100)
ax.grid(axis="x", alpha=0.25)

for bar, count in zip(bars, summary["count"]):
    ax.text(
        bar.get_width() + 2,
        bar.get_y() + bar.get_height() / 2,
        f"n={count}",
        va="center",
    )

fig.tight_layout()
path = Path("output/readable-summary.png")
path.parent.mkdir(parents=True, exist_ok=True)
fig.savefig(path, dpi=180, facecolor="white")
plt.close(fig)
print(path)

import json
from pathlib import Path
from datetime import datetime, timezone

manifest = {
    "input": "data/clean/study-clean.csv",
    "aggregation": "groupby topic: count, mean",
    "sort": "average_score descending",
    "axis": {"x": "topic", "y": "average_score", "y_unit": "0-100"},
    "filters": ["score is not missing", "0 <= score <= 100"],
    "generated_at": datetime.now(timezone.utc).isoformat(),
    "script": "plot_artifact.py",
}

path = Path("output/average-score.manifest.json")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
    json.dumps(manifest, ensure_ascii=False, indent=2),
    encoding="utf-8",
)
loaded = json.loads(path.read_text(encoding="utf-8"))
assert loaded["axis"]["y_unit"] == "0-100"
print("manifest keys:", sorted(loaded))

from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parents[1]
RAW_DIR = PROJECT_ROOT / "data" / "raw"
CLEAN_DIR = PROJECT_ROOT / "data" / "clean"
OUTPUT_DIR = PROJECT_ROOT / "output"

for directory in (RAW_DIR, CLEAN_DIR, OUTPUT_DIR):
    directory.mkdir(parents=True, exist_ok=True)

input_path = RAW_DIR / "study.csv"
clean_path = CLEAN_DIR / "study-clean.csv"
report_path = OUTPUT_DIR / "study-report.json"

print("root:", PROJECT_ROOT)
print("input exists:", input_path.exists())
print("clean output:", clean_path)
print("report output:", report_path)

import pandas as pd

def add_score_ratio(frame):
    data = frame.copy()
    data["score_ratio"] = data["score"] / 100
    return data

def summarize_by_topic(frame):
    return (
        frame.groupby("topic", as_index=False)
        .agg(
            records=("topic", "size"),
            average_score=("score", "mean"),
            total_minutes=("minutes", "sum"),
        )
        .sort_values("topic")
        .reset_index(drop=True)
    )

sample = pd.DataFrame({
    "topic": ["Python", "数据", "Python"],
    "score": [82, 91, 76],
    "minutes": [35, 50, 20],
})

with_ratio = add_score_ratio(sample)
summary = summarize_by_topic(with_ratio)

assert "score_ratio" not in sample.columns
assert summary["records"].sum() == len(sample)
print(with_ratio)
print(summary)

import json
from pathlib import Path
import pandas as pd

def make_sample():
    return pd.DataFrame({
        "topic": ["Python", "数据", "模型"],
        "minutes": [30, -5, 45],
        "score": [80, 90, 120],
    })

def split_valid_rows(frame):
    invalid = (
        frame["minutes"].lt(0)
        | frame["score"].lt(0)
        | frame["score"].gt(100)
    )
    return frame.loc[~invalid].copy(), frame.loc[invalid].copy()

sample = make_sample()
clean, rejected = split_valid_rows(sample)

assert len(clean) + len(rejected) == len(sample)
assert sample.equals(make_sample())
assert rejected["score"].max() == 120

payload = {
    "input_rows": len(sample),
    "cleaned_rows": len(clean),
    "rejected_rows": len(rejected),
}
path = Path("output/test-summary.json")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(payload), encoding="utf-8")
loaded = json.loads(path.read_text(encoding="utf-8"))
assert loaded["input_rows"] == loaded["cleaned_rows"] + loaded["rejected_rows"]
print("invariants passed")

import pandas as pd

sample = pd.DataFrame({
    "topic": ["Python", "数据", "模型"],
    "minutes": [30, -5, 45],
    "score": [80, 90, 120],
    "completed": [True, True, False],
})

from data_pipeline import aggregate, clean_input
clean, rejected = clean_input(sample)
assert len(clean) + len(rejected) == len(sample)
assert len(clean) == 1
assert len(rejected) == 2
assert aggregate(clean).loc[0, "topic"] == "Python"

from pathlib import Path
import pandas as pd

documents = pd.DataFrame([
    {
        "source_id": "guide-01",
        "version": "2026-08",
        "paragraph": 1,
        "text": "数组需要先确认 shape 和 dtype。",
        "license": "internal-note",
    },
    {
        "source_id": "guide-01",
        "version": "2026-08",
        "paragraph": 2,
        "text": "聚合结果应同时报告数量。",
        "license": "internal-note",
    },
])

records = documents.copy()
records["chunk_id"] = (
    records["source_id"] + ":" + records["paragraph"].astype(str)
)
records["char_count"] = records["text"].str.len()
records["split"] = "train"
records.loc[records["paragraph"] == 2, "split"] = "validation"

required = [
    "source_id", "version", "paragraph", "text",
    "license", "chunk_id", "char_count", "split",
]
assert set(required).issubset(records.columns)
assert records["chunk_id"].is_unique
assert records["char_count"].gt(0).all()
print(records[required])