1. КАК ИСПОЛЬЗОВАТЬ ЭТОТ ЭТАЛОН В ЛР2
1) Обучающийся копирует каркас, но обязан:
· добавить комментарии к ключевым функциям “что и зачем”,
· показать 3–5 строк данных “до/после” очистки,
· интерпретировать отчёт по пропускам и нарушениям,
· объяснить 2–3 принятых решения (например, почему заменили на None).
2) Отчёт по ЛР2 должен содержать раздел с ответами на вопросы:
· “Какие поля были наиболее проблемными?”
· “Какие правила качества чаще всего нарушались?”
· “Как это повлияет на анализ в NumPy (ЛР3)?”
2. ЭТАЛОННЫЙ КОД ЛР2 (PYTHON)
"""ЛР2. Работа с файлами и структурами данных. Первичная обработка набора данных Spotify.
Ожидаемая структура проекта:project/
│
├── data/
│ └── spotify_edu_dataset.csv
└── lab2_spotify_cleaning.py (или ноутбук lab2.ipynb)
Примечание:- В учебном датасете числовые поля намеренно сохранены как строки.- В ходе ЛР мы приводим типы, обрабатываем пропуски, выполняем первичную очистку и мини-анализ."""from __future__ import annotations
import csv
import re
from typing import Any, Optional
# -----------------------------# 1) Константы и настройки# ----------------------------- DATA_PATH = "data/spotify_edu_dataset.csv"
ENCODING = "utf-8"
# Поля, которые приводим к int / float
INT_FIELDS = ["duration_sec", "views", "likes", "comments"]
FLOAT_FIELDS = ["energy", "danceability", "tempo"]
# Поля для regex/текстаRELEASE_INFO_FIELD = "release_info"
DESC_FIELD = "description"
# Правила качества данных (для "углублённой части")# Возвращаем (ok: bool, message: str)def rule_energy_range(value: Optional[float]) -> bool:
return value is None or (0.0 <= value <= 1.0)
def rule_danceability_range(value: Optional[float]) -> bool:
return value is None or (0.0 <= value <= 1.0)
def rule_duration_positive(value: Optional[int]) -> bool:
return value is None or (value > 0)
def rule_non_negative(value: Optional[int]) -> bool:
return value is None or (value >= 0)
# -----------------------------# 2) Базовые утилиты приведения типов# -----------------------------def normalize_cell(value: Any) -> str:
"""
Нормализация значения из CSV:
- None -> ""
- прочее -> str + strip
"""if value is None:
return ""return str(value).strip()
ef to_int(value: Any) -> Optional[int]:
"""
Приведение к int:
- пустое -> None
- допускаем пробелы, запятые в числе: "1,234" -> 1234- ошибки -> None
"""s = normalize_cell(value)
if s == "" or s.lower() in {"none", "nan", "null"}:
return None s = s.replace(" ", "").replace(",", "")
try:return int(s)
except ValueError:
return None def to_float(value: Any) -> Optional[float]:
"""
Приведение к float:
- пустое -> None
- допускаем запятую как десятичный разделитель: "0,85" -> 0.85- ошибки -> None
"""s = normalize_cell(value)
if s == "" or s.lower() in {"none", "nan", "null"}:
return None s = s.replace(" ", "").replace(",", ".")
try:return float(s)
except ValueError:
return None
# -----------------------------# 3) Regex и текстовая обработка
# ----------------------------- YEAR_RE = re.compile(r"(\d{4})")
def extract_year(release_info: Any) -> Optional[int]:
"""
Извлекает год (4 цифры) из строки release_info.
Если год не найден или некорректен -> None.
"""
s = normalize_cell(release_info)
if not s:
return None m = YEAR_RE.search(s)
if not m:
return None year = to_int(m.group(1))
# Простая разумная проверка года (можно адаптировать)
if year is None or year < 1900 or year > 2100:
return Nonereturn year
NON_TEXT_RE = re.compile(r"[^\w\s]", flags=re.UNICODE)
def clean_text(text: Any) -> str:
"""
Очистка текста: - приводит к строке - удаляет спецсимволы (оставляет буквы/цифры/_ и пробелы) - схлопывает множественные пробелы - приводит к нижнему регистру (опционально, но полезно)"""
s = normalize_cell(text)
if not s:
return "" s = s.lower()
s = NON_TEXT_RE.sub(" ", s)
s = re.sub(r"\s+", " ", s).strip()
return s
# -----------------------------# 4) Чтение CSV и построение list[dict]
# ----------------------------- def read_csv_as_list_of_dicts(path: str, encoding: str = "utf-8") -> list[dict[str, str]]:
"""
Чтение CSV через стандартный модуль csv.
Возвращает список словарей: {поле: значение_строкой}."""
rows: list[dict[str, str]] = []
with open(path, "r", encoding=encoding, newline="") as f:
reader = csv.DictReader(f)
for row in reader:
# DictReader уже делает {header: cell}, но cell всё ещё строкаnormalized = {k: normalize_cell(v) for k, v in row.items()}
rows.append(normalized)
return rows
# -----------------------------# 5) Очистка записи (одной строки таблицы)# ----------------------------- def convert_types(row: dict[str, str]) -> dict[str, Any]:
"""
Возвращает новую запись, где числовые поля приведены к int/float,
а прочие поля остаются строками."""
new_row: dict[str, Any] = dict(row)
for field in INT_FIELDS:
new_row[field] = to_int(row.get(field, ""))
for field in FLOAT_FIELDS:
new_row[field] = to_float(row.get(field, ""))
return new_row
def add_derived_fields(row: dict[str, Any]) -> dict[str, Any]:
""" Добавляет вычисляемые поля: - release_year - description_clean - description_len """new_row = dict(row)
release_info = new_row.get(RELEASE_INFO_FIELD, "")
new_row["release_year"] = extract_year(release_info)
desc = new_row.get(DESC_FIELD, "")
desc_clean = clean_text(desc)
new_row["description_clean"] = desc_clean
new_row["description_len"] = len(desc_clean)
return new_row
# -----------------------------# 6) Правила качества данных + отчёты нарушений# ----------------------------- def validate_rules(row: dict[str, Any], violations: dict[str, int]) -> dict[str, Any]:
"""
Проверяет правила качества данных. Если нарушено – заменяет значение на None и увеличивает счётчик.
violations: словарь счётчиков нарушений по имени правила/поля.
"""
new_row = dict(row)
# energy in [0,1]energy = new_row.get("energy")
if not rule_energy_range(energy):
violations["energy_out_of_range"] = violations.get("energy_out_of_range", 0) + 1
new_row["energy"] = None
# danceability in [0,1]dance = new_row.get("danceability")
if not rule_danceability_range(dance):
violations["danceability_out_of_range"] = violations.get("danceability_out_of_range", 0) + 1
new_row["danceability"] = None
# duration_sec > 0duration = new_row.get("duration_sec")
if not rule_duration_positive(duration):
violations["duration_not_positive"] = violations.get("duration_not_positive", 0) + 1
new_row["duration_sec"] = None
# views/likes/comments >= 0 for field in ["views", "likes", "comments"]:
val = new_row.get(field)
if not rule_non_negative(val):
violations[f"{field}_negative"] = violations.get(f"{field}_negative", 0) + 1
new_row[field] = None
return new_row
def count_missing(data: list[dict[str, Any]]) -> dict[str, int]:
"""
Подсчёт пропусков (None или пустая строка) по каждому полю.
"""
missing: dict[str, int] = {}
if not data:
return missing
# Соберём все ключи, которые встречаются (на случай неполных строк)all_keys = set()
for row in data:
all_keys.update(row.keys())
for key in all_keys:
missing[key] = 0
for row in data:
for key in all_keys:
val = row.get(key, None)
if val is None:
missing[key] += 1
elif isinstance(val, str) and val.strip() == "":
missing[key] += 1
return missing
def print_top_missing(missing: dict[str, int], top_n: int = 10) -> None:
"""
Выводит поля с наибольшим числом пропусков."""
items = sorted(missing.items(), key=lambda x: x[1], reverse=True)
print("\n[Диагностика] Топ полей по числу пропусков:")
for k, v in items[:top_n]:
print(f" {k:<20} -> {v}")
# -----------------------------# 7) Мини-анализ без Pandas
# ----------------------------- def count_by_genre(data: list[dict[str, Any]]) -> dict[str, int]:
"""
Частотный словарь по жанрам. """counts: dict[str, int] = {}
for row in data:
genre = normalize_cell(row.get("genre", ""))
if not genre:
genre = "unknown"
counts[genre] = counts.get(genre, 0) + 1
return counts
def top_n_items(d: dict[str, int], n: int = 5) -> list[tuple[str, int]]:
"""
Сортировка словаря по значениям (убывание), возврат топ-N.
"""
return sorted(d.items(), key=lambda x: x[1], reverse=True)[:n]
def find_max_views_track(data: list[dict[str, Any]]) -> Optional[dict[str, Any]]:
"""
Находит запись с максимальными views (игнорирует None).
"""
best: Optional[dict[str, Any]] = None
best_views: int = -1
for row in data:
v = row.get("views")
if isinstance(v, int) and v > best_views:
best_views = v
best = row
return best
def mean_int_field(data: list[dict[str, Any]], field: str) -> Optional[float]:
"""
Среднее значение целочисленного поля (игнорирует None).
"""
s = 0
cnt = 0
for row in data:
val = row.get(field)
if isinstance(val, int):
s += val
cnt += 1
if cnt == 0:
return Nonereturn s / cnt
# -----------------------------# 8) Основной сценарий# ----------------------------- def main() -> None:
# 1) Чтение сырого файла
raw_data = read_csv_as_list_of_dicts(DATA_PATH, encoding=ENCODING)
print(f"[Шаг 1] Прочитано строк: {len(raw_data)}")
if raw_data:
print("\nПервые 3 записи (сырой вид, все поля строками):")
for i, row in enumerate(raw_data[:3], start=1):
print(f"\n--- Row {i} ---")
for k, v in row.items():
print(f"{k}: {v}")
# 2) Приведение типов + вычисляемые поля + правила качестваviolations: dict[str, int] = {}
clean_data: list[dict[str, Any]] = []
for row in raw_data:
r1 = convert_types(row)
r2 = add_derived_fields(r1)
r3 = validate_rules(r2, violations)
clean_data.append(r3)
print(f"\n[Шаг 2] Очистка выполнена. Итоговых строк: {len(clean_data)}")
# 3) Диагностика пропусковmissing = count_missing(clean_data)
print_top_missing(missing, top_n=12)
# 4) Отчёт по нарушениям правил качества
print("\n[Диагностика] Нарушения правил качества данных:")
if not violations:
print(" Нарушений не обнаружено.")
else:for k, v in sorted(violations.items(), key=lambda x: x[1], reverse=True):
print(f" {k:<25} -> {v}")
# 5) Мини-анализ без Pandas
genre_counts = count_by_genre(clean_data)
top_genres = top_n_items(genre_counts, n=5)
print("\n[Мини-анализ] Топ-5 жанров по числу треков:")
for g, c in top_genres:
print(f" {g:<15} -> {c}")
best = find_max_views_track(clean_data)
if best is not None:
print("\n[Мини-анализ] Трек с максимальными просмотрами:")
print(f" track_name: {best.get('track_name')}")
print(f" artist: {best.get('artist')}")
print(f" genre: {best.get('genre')}")
print(f" views: {best.get('views')}")
else:
print("\n[Мини-анализ] Невозможно определить трек с max views (нет данных views).")
avg_duration = mean_int_field(clean_data, "duration_sec")
if avg_duration is not None:
print(f"\n[Мини-анализ] Средняя длительность (сек): {avg_duration:.2f}")
else:
print("\n[Мини-анализ] Среднюю длительность вычислить нельзя (нет корректных duration_sec).")
# 6) Короткая проверка типов (диагностически) # (Показываем, что теперь числа – это числа, а не строки)if clean_data:
sample = clean_data[0]
print("\n[Контроль] Типы данных в первой очищенной записи:")
for key in ["duration_sec", "energy", "views", "release_year", "description_len"]:
print(f" {key:<15} -> {sample.get(key)} (type: {type(sample.get(key)).__name__})")
if __name__ == "__main__":
main()
3. СПЕЦИФИКАЦИЯ УЧЕБНОГО НАБОРА ДАННЫХ SPOTIFY_EDU_DATASET.CSV
Назначение
набора данных. Файл spotify_edu_dataset.csv является учебным набором данных для
лабораторной работы №2 и предназначен для формирования навыков углублённого
применения Python при подготовке данных:
чтение CSV, преобразование типов,
обработка пропусков, проверка корректности, применение регулярных выражений,
построение агрегатов без Pandas.
Формат. CSV (Comma-Separated Values), кодировка UTF-8. Первая строка – заголовок. Рекомендуемое размещение в проекте.
project/data/spotify_edu_dataset.csv
Объём. Около 300 строк (учебный объём).
Поля и типы (как записано в файле). Обратите внимание: ряд числовых полей хранится как строка, чтобы обучающийся выполнил приведение типов программно (таблица П3.1).
Таблица П3.1 – Поля и типы набора данных
|
Поле |
Тип в CSV |
Описание |
|
|
|
Название трека (русский формат |
|
artist |
|
Исполнитель ( |
|
genre |
|
Жанр (Поп, Рок, Хип-Хоп, Электронная, Инди, R&B) |
|
|
|
Текстовое поле с годом (например: «Выпущен в 2019 году») |
|
|
|
Длительность в секундах (требует |
|
energy |
|
Показатель 0–1 (требует |
|
|
|
Показатель 0–1 (требует |
|
tempo |
|
BPM (требует |
|
views |
|
Просмотры (требует |
|
likes |
|
Лайки (требует |
|
|
|
Комментарии (требует |
|
|
|
Описание (текст для очистки и regex) |
|
|
|
|
Намеренно внесённые особенности (учебная «грязь»). Набор данных включает контролируемые отклонения, имитирующие реальные данные:
1) Пропуски: часть значений views задана пустой строкой.
2) Некорректные диапазоны: часть значений energy может быть > 1.
3) Некорректные значения: часть значений duration_sec может быть отрицательной.
4) Сложные текстовые описания: примерно у 12% строк поле description содержит осложняющие
элементы для обработки регулярными выражениями, включая (не ограничиваясь):
·
даты
форматов dd.mm.yyyy и yyyy-mm-dd;
·
проценты
(85%), рейтинги (7/10), версии (ver.2.3, beta-4);
·
идентификаторы (ID:123456, код:AB12CD34);
·
ссылки
вида https://example.com/...;
·
теги (#музыка, #премьера), эмодзи, спецсимволы;
·
номера телефонов
вида +7(…);
·
таймкоды вида 01:35.
Методическая интерпретация. Файл позволяет отработать полный цикл первичной подготовки данных средствами Python:
«внешний файл → структурирование list[dict] → очистка → диагностика →
мини-анализ»,
а также подготовить основу для ЛР3 (NumPy) и ЛР4 (Pandas).
4. ПРИМЕР КОДА СОЗДАНИЯ УСЛОЖНЁННОЙ ВЕРСИИ УЧЕБНОГО ДАТАСЕТА SPOTIFY_EDU_DATASET.CSV
import
pandas as pd, random, re, string
path = "data/spotify_edu_dataset.csv"
df = pd.read_csv(path)
random.seed(42)
complex_templates = [
'🔥 ОФИЦИАЛЬНОЕ ВИДЕО №{n}: {tag} | премьера {dd}.{mm}.{yyyy} | ID:{id}',
'СМОТРИ ДО КОНЦА!!! {pct}% эмоций; рейтинг {rating}/10; ссылка: https://example.com/{slug}',
'Концерт (live) {yyyy} – трек "{track}" [ver.{ver}] (feat. {feat}) #топ{n} ###',
'Новый клип: "{track}" – {yyyy}. Длительность {dur}с. Просмотры: {views}. ❤{likes} 💬{com}',
'Текст: куплет1... припев!!! ({yyyy}) @{handle} $${money}k !!! ### {emoji}',
'Ремастер 20{yy} | EP-{ep} | код:{code} | жанр:{genre} | +7({phone})',
'Сезон {season}: серия {episode} – "{track}" | таймкод 01:{sec:02d} | [HD] [18+]',
'⚠️ Внимание: {track}! версия {ver} (beta-{beta}) – ошибки исправлены на {fix_date}',
'Плейлист: {n} треков; BPM={bpm}; key={key}; mood={mood}; #music #{genre_tag}',
'Комментарий автора: "{quote}" – {yyyy}-{mm}-{dd} (rev.{rev})'
]
tags = ["#музыка", "#новинка",
"#премьера", "#хит",
"#лайв", "#клип",
"#тренды"]
moods = ["happy", "sad", "energetic",
"chill", "romantic", "focus"]
keys = ["C", "Dm", "G", "Am",
"F#", "Bb"]
genre_tags = ["pop", "rock", "hiphop",
"electro", "indie", "rnb"]
emojis = ["🎵",
"🔥",
"✨",
"😎",
"💥",
"❤️",
"🎧"]
def slugify(s):
s = re.sub(r"\s+", "-", s.strip().lower())
s = re.sub(r"[^a-z0-9\-]", "", s)
return s[:20] if s else "link"
# choose ~12% rows to have complex description
idx = df.sample(frac=0.12, random_state=42).index.tolist()
for i in idx:
row = df.loc[i]
yyyy = random.randint(2000, 2023)
mm = random.randint(1, 12)
dd = random.randint(1, 28)
yy = str(yyyy)[-2:]
tpl = random.choice(complex_templates)
track = str(row.get("track_name", f"Трек_{i+1}"))
genre = str(row.get("genre", "Поп"))
desc = tpl.format(
n=random.randint(1, 99),
tag=random.choice(tags),
dd=f"{dd:02d}",
mm=f"{mm:02d}",
yyyy=yyyy,
yy=yy,
id=random.randint(100000, 999999),
pct=random.randint(50, 110),
rating=random.randint(1, 10),
slug=slugify(track),
track=track,
ver=f"{random.randint(1,3)}.{random.randint(0,9)}",
feat=f"Исполнитель_{random.choice(list(string.ascii_uppercase[:10]))}",
dur=random.randint(150, 320),
views=random.randint(10000, 500000000),
likes=random.randint(1000, 2000000),
com=random.randint(100, 100000),
handle="author_" + ''.join(random.choices(string.ascii_lowercase,
k=5)),
money=random.randint(1, 99),
emoji=random.choice(emojis),
ep=random.randint(1, 50),
code=''.join(random.choices(string.ascii_uppercase + string.digits,
k=8)),
genre=genre,
phone=str(random.randint(100,999)) +
str(random.randint(1000000,9999999)),
season=random.randint(1, 9),
episode=random.randint(1, 24),
sec=random.randint(0, 59),
beta=random.randint(1, 20),
fix_date=f"{random.randint(2020,2026)}-{random.randint(1,12):02d}-{random.randint(1,28):02d}",
bpm=random.randint(80, 180),
key=random.choice(keys),
mood=random.choice(moods),
genre_tag=random.choice(genre_tags),
quote="".join(random.choices(list("абвгдеёжзийклмнопрстуфхцчшщъыьэюя ") +
list("0123456789!?.,:-"), k=40)).strip(),
rev=random.randint(1, 15),
)
df.at[i, "description"] = desc
out_path = "data/spotify_edu_dataset.csv"
df.to_csv(out_path, index=False, encoding="utf-8")
# приведем несколько сложных примеров
sample_complex
= df.loc[idx, ["track_name", "genre", "release_info",
"description"]].head(5)
(out_path, sample_complex)
Результат
('data/spotify_edu_dataset.csv',
track_name ... description
203 Трек_204 ... Текст: куплет1... припев!!! (2020) @author_cgp...
266 Трек_267 ... ⚠️ Внимание: Трек_267! версия 1.3 (beta-4) – о...
152 Трек_153 ... Текст: куплет1... припев!!! (2023) @author_aop...
9 Трек_10 ... Комментарий автора: "х..аюыдщшёрглнфъо-6щ ъ.е9...
233 Трек_234 ... Концерт (live) 2000 – трек "Трек_234" [ver.3.6...
[5 rows x 4 columns])