1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@engchina

【Python】文字化け・フォーマットエラーにさようなら！実務で使える堅牢なデータアップロード処理の実装

1
Posted at

はじめに

Webアプリケーションやデータ分析基盤を開発していると、ユーザーから「CSVファイルをアップロードして分析する機能」を求められることがよくあります。
しかし、一口に「CSV」や「データファイル」と言っても、実態はカオスです。

  • Excelで保存したCSV（Shift_JIS / CP932）
  • モダンなエディタで保存したCSV（UTF-8）
  • BOM付きのUTF-8（UTF-8-SIG）
  • 実はタブ区切り（TSV）だった
  • JSONファイルだけど、配列だったり辞書だったり、JSON Linesだったりする

これら全てに対応しようとすると、単純な pd.read_csv() だけではエラーや文字化けが頻発します。
本記事では、私が実際のプロジェクト（機械学習プラットフォーム）で実装した、「とりあえず投げればよしなに読み込んでくれる」 堅牢なデータローダーのロジックを解説・共有します。

1. 文字コードの壁を突破する（エンコーディング自動判定）

日本国内の業務システムでは、依然として Shift_JIS (CP932) が現役です。一方でWeb標準は UTF-8 です。これらを自動で判別する必要があります。
chardet などのライブラリもありますが、処理速度と精度のバランスを考えると、「可能性の高い順にデコードを試行する」 戦略が最も実用的でした。

実装コード: detect_encoding 

from typing import Tuple

def detect_encoding(content: bytes) -> Tuple[str, str]:
    """
    バイトデータのエンコーディングを検出してテキストに変換する
    
    Args:
        content: ファイルのバイトデータ
        
    Returns:
        (decoded_text, encoding_name): デコード済みテキストとエンコーディング名
    """
    # 日本の環境でよく遭遇するエンコーディングの優先順位リスト
    encodings = [
        "utf-8",
        "utf-8-sig",  # BOM付きUTF-8（Excel出力などで多い）
        "shift_jis",
        "cp932",      # Windows拡張Shift_JIS（はしご高などが含まれる場合）
        "euc_jp",
        "iso-2022-jp",
        "latin1"      # 最終手段
    ]
    
    for encoding in encodings:
        try:
            decoded = content.decode(encoding)
            return decoded, encoding
        except (UnicodeDecodeError, LookupError):
            continue
    
    # すべて失敗した場合はlatin1で強制デコード（エラー回避）
    return content.decode("latin1", errors="replace"), "latin1"

ポイント:

  • utf-8-sig をリストに含めることで、BOM（Byte Order Mark）付きファイルも正しく扱えます。
  • cp932 を shift_jis の後に配置（あるいは代替）することで、Windows固有文字への対応を強化できます。

2. CSVとTSVの境界を曖昧にする（区切り文字の自動判定）

拡張子が .csv でも中身がタブ区切りだったり、その逆だったりすることは日常茶飯事です。
Pandasの read_csv には強力な自動判定機能がありますが、明示的に使うにはコツがいります。

実装ロジック

import io
import pandas as pd

def load_csv_or_tsv(text: str, filename: str) -> pd.DataFrame:
    # TSVの判定（拡張子または内容にタブが含まれるか）
    if filename.endswith('.tsv') or '\t' in text[:1000]:
        df = pd.read_csv(io.StringIO(text), sep='\t')
    else:
        # Pythonエンジンを指定し、sep=Noneにすることで区切り文字をスニッフィングさせる
        df = pd.read_csv(io.StringIO(text), sep=None, engine='python')
    
    # 【重要】列名のBOM除去
    # ファイル自体のエンコーディングがUTF-8でも、列名文字列の先頭にBOM(U+FEFF)が
    # 残ってしまうケースがあるため、明示的に除去します。
    df.columns = [str(col).lstrip('\ufeff') for col in df.columns]
    
    return df

ポイント:

  • sep=None, engine='python' を指定することで、Pandasが区切り文字（カンマ、タブ、セミコロン等）を推論してくれます。
  • lstrip('\ufeff') は地味ですが重要です。これがないと df['ID'] でアクセスしようとしたら KeyError: '\ufeffID' と言われる幽霊バグに悩まされます。

3. どんなJSONでもDataFrameにしたい（構造の正規化）

JSONは自由度が高すぎるため、pd.read_json() 一発ではうまくいかないことが多いです。

  • リスト形式: [{"col": 1}, {"col": 2}]
  • 辞書形式: {"0": {"col": 1}, "1": {"col": 2}}
  • ネスト: {"data": [{"col": 1}, ...]}
  • JSON Lines: 改行区切りのJSONオブジェクト

これらを総当たりで解析し、フラットな表形式（DataFrame）に変換するロジックです。

実装コード: _load_json 

import json

def load_flexible_json(text: str) -> pd.DataFrame:
    """あらゆる形式のJSONテキストをDataFrameに変換を試みる"""
    
    # 1. 通常のJSONとしてパース
    try:
        data = json.loads(text)
        
        # パターンA: リスト形式 [{}, {}] -> 素直に変換
        if isinstance(data, list):
            return pd.json_normalize(data) if data and isinstance(data[0], dict) else pd.DataFrame({"value": data})
        
        # パターンB: 辞書形式
        elif isinstance(data, dict):
            # B-1: 行インデックス形式 {"0": {...}, "1": {...}}
            if data and all(isinstance(v, dict) for v in data.values()):
                return pd.DataFrame.from_dict(data, orient='index')
            
            # B-2: 一般的なキーにデータが包まれている場合 {"data": [...], "meta": ...}
            # よくあるキー名を探索
            for key in ["data", "rows", "items", "records", "results"]:
                if key in data and isinstance(data[key], list):
                    return pd.json_normalize(data[key])
            
            # B-3: 単一オブジェクトの場合 -> 1行のデータとして扱う
            return pd.json_normalize(data)
            
    except json.JSONDecodeError:
        pass
    
    # 2. JSON Lines形式（改行区切りJSON）としてパース
    try:
        items = []
        for line in text.splitlines():
            line = line.strip()
            if line:
                try:
                    items.append(json.loads(line))
                except:
                    continue
        if items:
            return pd.json_normalize(items)
    except:
        pass
        
    raise ValueError("サポートされていないJSON形式です")

完全なコード例（コピペ用）

これらをまとめたクラスです。FastAPIの UploadFile を受け取る想定ですが、通常のファイルパスやバイト列にも容易に応用可能です。

import io
import json
import pandas as pd
from typing import Tuple, Optional, Any
from fastapi import UploadFile, HTTPException

class UniversalDataLoader:
    @staticmethod
    def detect_encoding(content: bytes) -> Tuple[str, str]:
        encodings = ["utf-8", "utf-8-sig", "shift_jis", "cp932", "euc_jp", "iso-2022-jp", "latin1"]
        for encoding in encodings:
            try:
                return content.decode(encoding), encoding
            except (UnicodeDecodeError, LookupError):
                continue
        return content.decode("latin1", errors="replace"), "latin1"

    @classmethod
    def load_file(cls, file: UploadFile) -> Tuple[pd.DataFrame, str, str]:
        """
        メイン処理
        Returns: (DataFrame, encoding, file_type)
        """
        try:
            content = file.file.read()
            filename = file.filename.lower()
            
            # 1. エンコーディング判定
            text, encoding = cls.detect_encoding(content)
            
            df = pd.DataFrame()
            file_type = 'unknown'
            
            # 2. 形式に応じた読み込み
            try:
                # JSON判定 (拡張子 または 内容の先頭文字)
                if filename.endswith('.json') or text.strip().startswith(('{', '[')):
                    df = cls._load_json(text)
                    file_type = 'json'
                # TSV判定
                elif filename.endswith('.tsv') or '\t' in text[:1000]:
                    df = pd.read_csv(io.StringIO(text), sep='\t')
                    file_type = 'tsv'
                # CSV (デフォルト)
                else:
                    df = pd.read_csv(io.StringIO(text), sep=None, engine='python')
                    file_type = 'csv'
            except Exception as e:
                raise ValueError(f"パースエラー: {str(e)}")

            # 3. 後処理: 列名のBOM除去と文字列化
            df.columns = [str(col).lstrip('\ufeff') for col in df.columns]
            
            if df.empty:
                raise ValueError("データが空です")
                
            return df, encoding, file_type

        except Exception as e:
            # 実務ではここでロギングなどを行う
            raise HTTPException(status_code=400, detail=f"ファイル読み込み失敗: {str(e)}")

    @staticmethod
    def _load_json(text: str) -> pd.DataFrame:
        # (前述のJSON読み込みロジックと同様)
        try:
            data = json.loads(text)
            if isinstance(data, list):
                if data and isinstance(data[0], dict):
                    return pd.json_normalize(data)
                return pd.DataFrame({"value": data})
            elif isinstance(data, dict):
                if data and all(isinstance(v, dict) for v in data.values()):
                    return pd.DataFrame.from_dict(data, orient='index')
                for key in ["data", "rows", "items", "records"]:
                    if key in data and isinstance(data[key], list):
                        return pd.json_normalize(data[key])
                return pd.json_normalize(data)
        except:
            pass
            
        # JSON Lines fallback
        items = [json.loads(line) for line in text.splitlines() if line.strip()]
        if items:
            return pd.json_normalize(items)
            
        raise ValueError("JSON形式を解析できませんでした")

使い方

FastAPIのエンドポイントでの使用例です。

from fastapi import FastAPI, UploadFile, File

app = FastAPI()

@app.post("/upload")
async def upload_dataset(file: UploadFile = File(...)):
    loader = UniversalDataLoader()
    df, encoding, ftype = loader.load_file(file)
    
    return {
        "filename": file.filename,
        "detected_encoding": encoding,
        "file_type": ftype,
        "columns": list(df.columns),
        "row_count": len(df),
        "preview": df.head(5).to_dict(orient="records")
    }

まとめ

ユーザーデータのアップロード機能は、一度作ってしまえば多くのプロジェクトで使い回せる資産になります。
今回のコードは以下の点を考慮しています。

  1. Shift_JIS / UTF-8 の混在環境に対応
  2. CSV / TSV の曖昧さを吸収
  3. JSON の多様な構造に対応
  4. BOM などの隠れたトラブル要因を除去

これをベースに、必要に応じてExcel (pd.read_excel) の対応などを追加していけば、さらに強力なインポーターになるでしょう。

1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?