Czytanie plików w Pythonie: tekst, CSV i kontrola importu

Czytanie pliku w Pythonie potrafi zająć jedną linijkę. Ustalenie, co właściwie zostało wczytane, bywa ciekawsze. Identyfikator traci zera, polskie znaki wyglądają podejrzanie, a pusta komórka nagle staje się zerem. Program nie zgłasza błędu, więc łatwo uznać sprawę za zamkniętą. Ja wolę zatrzymać się przy imporcie i sprawdzić kilka rzeczy, zanim z tych danych powstanie raport.

Przejdę od małego pliku tekstowego do CSV z kontrolą kolumn i typów. Wszystkie dane są syntetyczne. Przykłady korzystają z biblioteki standardowej Pythona 3.14; na końcu pokazuję również wariant pandas, sprawdzony w wersji 3.0.6. Bloki wykonuj kolejno w jednym skrypcie. Nie potrzebujesz pobierać Titanica ani szukać pliku, który autor tutoriala miał kiedyś na pulpicie.

Najpierw ustalam, gdzie Python szuka pliku

Ścieżka względna jest liczona od bieżącego katalogu roboczego procesu. Nie musi to być folder, w którym leży skrypt. Edytor, notebook i terminal mogą uruchamiać ten sam kod z innych miejsc. Jeżeli dostajesz FileNotFoundError, zacznij od sprawdzenia katalogu roboczego i nazwy pliku, zamiast zmieniać przypadkowe ustawienia kodowania.

Do budowania ścieżek wybieram Path z modułu pathlib. Łączenie folderów operatorem ukośnika ogranicza pomyłki z separatorami. Zwykły napis zawierający \t oznacza tabulator, a \n znak nowej linii. To istotne przy kopiowaniu windowsowych ścieżek. Możesz użyć surowego napisu albo Path, pamiętając, że sam Path nie cofnie wcześniejszej interpretacji znaków w literale.

Ćwiczenie umieszczam w osobnym katalogu tymczasowym. Dzięki temu powstają tylko nasze próbki i nie nadpisujemy przypadkiem prawdziwego eksportu. Obiekt workspace zostawiam do końca ćwiczenia. Ostatni blok usunie utworzony katalog wraz z plikami. W normalnym projekcie w miejsce tego katalogu podajesz uzgodnioną lokalizację danych.

# example: workspace
from pathlib import Path
from tempfile import TemporaryDirectory

workspace = TemporaryDirectory(prefix="file-reading-")
root = Path(workspace.name)
text_path = root / "notes.txt"
text_path.write_text("Łódź\nKraków\n", encoding="utf-8")
print(text_path.name, text_path.is_file())

Wynik to notes.txt True. Tworzenie próbki jest częścią przygotowania zadania, a dalszy odczyt nie modyfikuje pliku. Dokumentacja pathlib opisuje operacje na ścieżkach i plikach. Nie traktuję jednak samego is_file() jako gwarancji późniejszego odczytu: plik może zniknąć albo proces może nie mieć wymaganych uprawnień.

Otwieram tekst przez with i jawne kodowanie

Funkcja open zwraca obiekt pliku. W trybie r odczytuję tekst, a w rb odczytuję bajty. W tym ćwiczeniu potrzebuję tekstu. Podaję encoding="utf-8", ponieważ w takim kodowaniu utworzyłem próbkę. To konkretna informacja o pliku, a nie magiczna opcja naprawiająca każdy eksport. Starszy plik może wymagać innego, potwierdzonego kodowania.

# example: text
with text_path.open("r", encoding="utf-8") as handle:
    contents = handle.read()
    print(repr(contents))
    print(handle.closed)
print(handle.closed)

Najpierw zobaczysz 'Łódź\nKraków\n', następnie False i True. repr pokazuje znaki końca linii, które zwykły print zamieniłby na przejście do kolejnego wiersza. Wewnątrz bloku plik pozostaje otwarty, po wyjściu jest zamknięty. Tak samo zostanie zamknięty, jeśli kod w bloku zakończy się wyjątkiem. Zmienna contents nadal przechowuje odczytany tekst.

Przy ręcznym open i close łatwo pominąć zamknięcie w jednej ze ścieżek programu. Menedżer kontekstu porządkuje ten obowiązek. W przewodniku Pythona znajdziesz również krótszy wariant odczytu małego pliku. Dla nauki wybieram tutaj blok with, bo wyraźnie pokazuje czas korzystania z zasobu.

Nie ustawiam odruchowo errors="ignore", gdy pojawi się błąd dekodowania. Taka opcja może usunąć fragmenty tekstu i zostawić pozornie poprawne nazwy albo identyfikatory. Najpierw sprawdzam pochodzenie pliku i jego kodowanie. Jeśli świadomie dopuszczam zastępowanie błędnych znaków, opisuję ten kompromis i kontroluję, które rekordy zmieniły treść.

read, readline i iteracja mają różne zastosowania

read() bez argumentu pobiera całą pozostałą zawartość. Przy małej notatce to wygodne. Przy dużym pliku może niepotrzebnie zajmować pamięć. readline() pobiera pojedynczą linię, a iteracja po obiekcie pliku pozwala przechodzić przez kolejne linie. Trzy wywołania readline() nie stają się automatycznie importem całego zbioru, nawet jeśli próbka miała akurat trzy wiersze.

# example: lines
with text_path.open(encoding="utf-8") as handle:
    first = handle.readline()
    remaining = list(handle)
    after_end = handle.read()
print(first.rstrip("\n"))
print([line.rstrip("\n") for line in remaining])
print(repr(after_end))

Wynik to Łódź, lista ['Kraków'] i pusty napis ''. Odczyt przesuwa bieżącą pozycję w pliku. Po dojściu do końca kolejne czytanie nie zacznie samo od początku. Jeśli potrzebuję ponownie przejść przez dane, otwieram plik jeszcze raz albo świadomie ustawiam pozycję. Dla początkującego ponowne otwarcie bywa najczytelniejszym rozwiązaniem.

Używam rstrip("\n"), aby usunąć znak końca linii. Zwykłe strip() usuwa także inne białe znaki z obu stron, co może zmienić znaczenie pól o stałej szerokości. Dobieraj czyszczenie do formatu. W logu końcowa spacja bywa bez znaczenia, ale w surowym materiale do diagnozy właśnie ona może wyjaśniać problem.

W przykładzie zamieniam pozostałe linie na listę, żeby łatwo pokazać wynik. Przy dużym pliku przetwarzaj linię bezpośrednio w pętli i nie zbieraj wszystkiego do kolejnej listy. Oszczędność pamięci wynika z całego sposobu przetwarzania. Samo użycie pętli nie wystarczy, jeśli i tak odkładasz każdą linię w pamięci.

CSV czytam parserem, nie dzieleniem po przecinku

CSV jest tekstem z regułami zapisu rekordów. Separator może znajdować się wewnątrz pola ujętego w cudzysłowy. Takie pole może zawierać również znak nowej linii. Dlatego split(",") nie zastępuje parsera CSV. Przykład z dwoma prostymi kolumnami może działać, a następnego dnia komentarz klienta rozbije liczbę pól.

Przygotujmy trzy rekordy syntetycznej kolejki zadań. task_id jest identyfikatorem, city nazwą, minutes czasem, a note dowolnym opisem. Zero minut oznacza tu poprawny pomiar zera. Puste pole oznacza brak pomiaru. Te dwa przypadki celowo występują obok siebie, bo importer powinien zachować ich różne znaczenia.

# example: csv-fixture
import csv

csv_path = root / "tasks.csv"
with csv_path.open("w", encoding="utf-8", newline="") as handle:
    writer = csv.writer(handle, delimiter=";")
    writer.writerow(["task_id", "city", "minutes", "note"])
    writer.writerows([
        ["001", "Łódź", "12", "call; follow-up"],
        ["002", "Kraków", "0", ""],
        ["003", "Gdańsk", "", "two\nlines"],
    ])

Separator to średnik, ale pierwsza notatka również zawiera średnik. csv.writer prawidłowo umieści ją w cudzysłowach. Ostatnia notatka zajmuje dwie linie tekstu, choć nadal należy do jednego rekordu. Liczenie fizycznych linii pliku nie da więc liczby zadań. Licz rekordy zwracane przez parser, a nagłówek traktuj zgodnie z formatem.

Parametr newline="" pozostawia obsługę końców linii modułowi csv. To zalecenie z dokumentacji csv, ważne także przy cytowanych polach wielowierszowych. Nie oznacza usuwania końców linii z danych. Kodowanie i separator opisują różne aspekty pliku; poprawienie jednego nie rozwiązuje błędu drugiego.

Zachowuję identyfikator i sprawdzam nagłówek

DictReader wykorzystuje pierwszy rekord jako nazwy pól i daje dla kolejnych rekordów słowniki. Dzięki temu odwołuję się do task_id, a nie do tajemniczej kolumny numer zero. Zwykły odczyt modułem csv pozostawia wartości tekstowe. To dobre miejsce, by świadomie zdecydować, które kolumny zamienić na liczby.

# example: csv-read
with csv_path.open(encoding="utf-8", newline="") as handle:
    reader = csv.DictReader(handle, delimiter=";")
    expected_columns = ["task_id", "city", "minutes", "note"]
    if reader.fieldnames != expected_columns:
        raise ValueError("Unexpected columns")
    rows = list(reader)
print(len(rows), rows[0]["task_id"])
print(rows[0]["note"])
print(repr(rows[2]["note"]))

Dostaniesz 3 001, potem call; follow-up oraz 'two\nlines'. Identyfikator zachował zera. Importer nie pomylił średnika w notatce z separatorem kolumn i nie rozbił wielowierszowej notatki na dwa zadania. Te drobne kontrole mówią o jakości odczytu znacznie więcej niż samo wyświetlenie pierwszego rekordu.

Porównanie listy nagłówków wymaga tutaj konkretnej kolejności. To celowy kontrakt ćwiczenia. Jeśli system źródłowy może zmieniać kolejność kolumn, sprawdzaj uzgodniony zestaw nazw i ich unikalność. Nie akceptuj jednak dowolnych nagłówków tylko dlatego, że większość raportu jeszcze się liczy. Zdublowana nazwa kolumny może spowodować utratę informacji przy reprezentacji słownikowej.

Konwersję wartości łączę z kontrolą znaczenia

Samo int() nie wie, czy ujemna liczba minut ma sens w naszym zadaniu. Nie wie też, czy pusty napis oznacza zero, brak pomiaru czy wadliwy eksport. W ćwiczeniu ustalam zasady jawnie: identyfikator musi istnieć, liczba minut jest nieujemną liczbą całkowitą, pusty czas pozostaje None. Nie zgaduję wartości za źródło.

# example: validation
validated = []
for record_number, row in enumerate(rows, start=1):
    if None in row or any(value is None for value in row.values()):
        raise ValueError(f"Wrong field count in record {record_number}")
    if not row["task_id"]:
        raise ValueError(f"Missing task_id in record {record_number}")
    raw_minutes = row["minutes"]
    minutes = None if raw_minutes == "" else int(raw_minutes)
    if minutes is not None and minutes < 0:
        raise ValueError(f"Negative minutes in record {record_number}")
    validated.append({**row, "minutes": minutes})

known = [row["minutes"] for row in validated if row["minutes"] is not None]
print(len(validated), sum(known), len(validated) - len(known))

Wynik 3 12 1 oznacza trzy zadania, dwanaście znanych minut i jeden brak pomiaru. Numer w komunikacie błędu opisuje rekord danych, a nie linię pliku. To rozróżnienie przydaje się właśnie przy wielowierszowym CSV. Sprawdzenie None in row wykrywa dodatkowe pola, a wartości None ujawniają pola brakujące w zbyt krótkim rekordzie.

Odróżniam brak liczby od wartości fałszywej w sensie logicznym. Warunek if row["minutes"] po konwersji odrzuciłby zarówno None, jak i poprawne zero. Dlatego filtr zawiera is not None. Więcej o takich różnicach piszę w tekście o zmiennych, typach i operatorach.

W małym ćwiczeniu błąd zatrzymuje przetwarzanie. W większym procesie możesz oddzielić rekordy poprawne od odrzuconych, ale wtedy raport musi pokazywać ich liczby i przyczyny. Ciche pomijanie problematycznych wierszy tworzy wygodną iluzję kompletności. Nie ma uniwersalnej odpowiedzi, czy lepiej zatrzymać cały import; jest za to obowiązek wiedzieć, co z niego wypadło.

Kiedy przechodzę do pandas.read_csv

Jeśli kolejnym krokiem jest filtrowanie tabeli, grupowanie i łączenie danych, DataFrame bywa wygodniejszy od listy słowników. Używam wtedy pandas.read_csv, nadal jawnie podając separator, kodowanie i typy. Parametry odczytu powinny wynikać z kontraktu danych. Automatyczne rozpoznanie typu jest wygodne przy rozpoznaniu pliku, ale nie zawsze właściwe w powtarzanym raporcie.

# example: pandas-csv
import pandas as pd

df = pd.read_csv(
    csv_path,
    sep=";",
    encoding="utf-8",
    dtype={"task_id": "string", "city": "string", "minutes": "Int64", "note": "string"},
    keep_default_na=False,
    na_values={"minutes": [""]},
)
print(df["task_id"].tolist())
print(int(df["minutes"].sum()), int(df["minutes"].isna().sum()))
print(df.loc[1, "note"] == "")

Wyniki to ['001', '002', '003'], 12 1 i True. Typ Int64 z wielkiej litery obsługuje brakujące liczby całkowite. Wyłączam domyślną listę oznaczeń braków i wskazuję pusty tekst tylko dla kolumny minutes. Pusta notatka pozostaje pustą notatką. Dzięki temu importer nie rozciąga jednej reguły na wszystkie pola tabeli.

Nie jest to jedyne poprawne ustawienie. W innym źródle pusty tekst może oznaczać brak także w kolumnie miasta. Ustal to przed importem. Dokumentacja read_csv opisuje dtype, keep_default_na, na_values i odczyt partiami. Podanie dtype nie zastępuje kontroli ujemnego czasu, unikalności identyfikatora ani zgodności liczby rekordów ze źródłem.

Starsza wersja tego artykułu pokazywała np.recfromcsv. Funkcję usunięto w NumPy 2.0, co opisuje przewodnik migracji NumPy. Przy tablicach numerycznych można rozważyć genfromtxt. W tym zadaniu, z tekstowymi identyfikatorami i notatkami, wybieram csv lub pandas. Nie dokładam trzeciej biblioteki tylko po to, żeby otworzyć ten sam plik.

Duży plik i nieudany import wymagają osobnych decyzji

Odczyt całego CSV do listy albo całej tabeli do DataFrame wymaga pamięci na wynik. Rozmiar w pamięci może znacznie różnić się od rozmiaru pliku na dysku. Przy większym zbiorze rozważ iterację rekordów lub chunksize w pandas. Trzeba jednak ustalić, które obliczenia można wykonać partiami i jak połączyć wyniki bez zmiany znaczenia.

Sumy i liczby rekordów da się dodawać między partiami. Ze średnią trzeba zachować licznik i mianownik; średnia ze średnich partii może być błędna. Sprawdzanie duplikatów wymaga pamiętania identyfikatorów także z wcześniejszych partii. Dzielenie wejścia na kawałki rozwiązuje część problemu z pamięcią, ale nie projektuje za Ciebie całej analizy.

Przy awarii rozdzielam kilka pytań: czy plik istnieje, czy można go otworzyć, czy bajty dają się zdekodować, czy parser rozumie strukturę i czy wartości spełniają reguły. FileNotFoundError, UnicodeDecodeError i błąd konwersji liczby mówią o innych etapach. Jedno szerokie except, które drukuje „coś poszło nie tak”, zabiera informację potrzebną do naprawy.

Co sprawdzam przed ponownym użyciem importera

Przy kolejnym eksporcie porównuję nie tylko nazwy kolumn. Sprawdzam też okres danych, liczbę rekordów i to, czy plik jest pełnym zrzutem, czy przyrostem od poprzedniego uruchomienia. Dwa pliki o identycznym nagłówku mogą opisywać różne zakresy. Dodanie ich do siebie bez tej wiedzy może podwoić wynik. Sama funkcja odczytująca plik nie ma jak rozpoznać takiej pomyłki.

Osobno sprawdzam, czy identyfikator zadania ma być unikalny. W naszym małym przykładzie tak go interpretuję, ale kod walidacji nie wymusza jeszcze tej reguły. Możesz dopisać kontrolę zbioru już widzianych identyfikatorów i porównać jego rozmiar z liczbą rekordów. Jeśli ten sam identyfikator oznacza kolejne zdarzenia jednego zadania, usunięcie powtórzeń byłoby błędem. Najpierw określ, co reprezentuje pojedynczy wiersz.

Czasem pierwszy nagłówek wygląda poprawnie, a program zgłasza inną nazwę. Przyczyną może być znacznik BOM na początku pliku UTF-8. Do potwierdzonego takiego eksportu można użyć kodowania utf-8-sig, które obsługuje ten znacznik podczas odczytu. Nie naprawiam problemu przez ręczne obcinanie pierwszego znaku każdego nagłówka. Taka poprawka uszkodzi zwykły plik, gdy producent przestanie dodawać BOM.

Patrzę również na rzeczywistą zawartość, a nie wyłącznie rozszerzenie. Plik nazwany .csv może zawierać stronę błędu pobraną z serwera. Plik .txt może być poprawną tabelą z tabulatorami. W razie wątpliwości oglądam krótki fragment jako tekst i porównuję go z uzgodnionym formatem. Zmiana rozszerzenia pliku nie konwertuje jego struktury ani kodowania.

Warto zapisać oczekiwania obok kodu: separator, kodowanie, wymagane kolumny, znaczenie pustych pól i minimalne kontrole wyniku. Gdy ktoś dostanie ten skrypt za miesiąc, powinien rozumieć, dlaczego 001 zostaje tekstem i dlaczego jeden pomiar pozostał nieznany. To także ułatwia rozmowę z osobą przygotowującą eksport. Zamiast zgłaszać ogólnie „Python nie czyta pliku”, można wskazać konkretny rekord, pole i naruszoną zasadę.

Ćwiczenie: zepsuj dane i zobacz, co zatrzyma import

Zapisz przewidywany wynik przed uruchomieniem kodu. Potem porównaj go z rzeczywistym rezultatem i wyjaśnij każdą różnicę na danych wejściowych.

Zmień w próbce 12 na twelve, a potem osobno na -1. Pierwszy wariant zatrzyma konwersja na liczbę, drugi jawna reguła nieujemności. Następnie usuń identyfikator i dodaj dodatkowe pole w rekordzie. Każdy problem powinien zostać wykryty przed obliczeniem sumy. Uruchamiaj cały skrypt od początku po każdej zmianie, żeby nie korzystać z danych pozostawionych w poprzedniej sesji.

Sprawdź też poprawne przypadki graniczne: pustą notatkę, czas równy zero i plik zawierający sam nagłówek. Ten ostatni ma poprawną strukturę, ale zero rekordów. Czy to dopuszczalne w raporcie dziennym, zależy od procesu. Warto odróżnić „nie było zadań” od „nie dostaliśmy eksportu”. Tego nie ustali za nas funkcja odczytująca CSV.

# example: cleanup
workspace.cleanup()
print(root.exists())

Na końcu zobaczysz False: katalog ćwiczenia został usunięty. Masz teraz mały import, którego zachowanie można wyjaśnić na konkretnych rekordach. Następnym krokiem może być wybieranie fragmentów listy albo pełny pierwszy raport w pandas. Ja najpierw upewniam się, że odczytane dane znaczą to samo co dane zapisane w źródle.