pythonちょい嵌り-'shift_jis' codec can't encode character '\uff5e' in position 1: illegal multibyte sequence

はじめに

Django2で動的にformを作る2-MultipleChoiceFieldに対応させる-で作成したものを加工して
CSV形式でZIPファイルに出力するものを作りました。

クライアント端末であるWindows10から入力した「～」文字が
CSVで吐き出す際にUnicodeからShift-JISにマッピングできず困惑しました。
出力されるエラーはこんな感じ
'shift_jis' codec can't encode character '\uff5e' in position 1: illegal multibyte sequence

これも嵌りポイントだと思うので記事にして行きます(^^♪

お試しコード

動的にformを作る2のソースにcsv,zip出力を組み込んだものとなります。

urls.py

from django.urls import path

from . import views

urlpatterns = [
    path('', views.dynamic, name='index'),
]

forms.py

from django import forms

class EnqForm(forms.Form):
    pass

views.py

from django.shortcuts import render
from .forms import EnqForm
from django import forms
from django.http import HttpResponse
import io
import os
import csv, zipfile
import tempfile

# Create your views here.
def dynamic(request):
    context = {}
    content = {}
    form_item = {}
    qs = []

    # define enquete fields
    qs.append( {'title':'title1', 'description': 'note1', 'type': 'text' , 'required': False} )
    qs.append( {'title':'title2', 'description': 'note2', 'type': 'text' , 'required': True} )
    qs.append( {'title':'title3', 'description': 'note3', 'type': 'radio' , 'required': False} )
    qs.append( {'title':'title4', 'description': 'note4', 'type': 'multi' , 'required': True} )
    qs.append( {'title':'title5', 'description': 'note5', 'type': 'radio' , 'required': True} )

    # create enquete form objects
    no = 0
    for q in qs:
        if q['type'] == 'text':
            form_item.update( { ('ans%d' % no): forms.CharField(label=q['title'], label_suffix=q['description'], required=q['required'], max_length=256) } )
        elif q['type'] == 'radio':
            form_item.update( { ('ans%d' % no): forms.ChoiceField(label=q['title'], label_suffix=q['description'], required=q['required'], choices=(('a','a'),('b','b'),('c','c'),('d','d')), widget=forms.RadioSelect) } )
        elif q['type'] == 'multi':
            form_item.update( { ('ans%d' % no): forms.MultipleChoiceField(label=q['title'], label_suffix=q['description'], required=q['required'], choices=(('a','a'),('b','b'),('c','c'),('d','d')), widget=forms.CheckboxSelectMultiple) } )
        no += 1

    # create dynamic form
    DynamicEnqForm = type('DynamicEnqForm', (EnqForm,), form_item )

    # get enquete answers
    if request.method == 'POST':
        formset = DynamicEnqForm(request.POST or None)
        if formset.is_valid():
            for key in formset.cleaned_data:
                if key != 'csrfmiddlewaretoken' and key != 'method':
                    content[key] = formset.cleaned_data[key]
            if 'method' in request.POST and request.POST['method'] == 'regist':
                # regist

                fp = tempfile.NamedTemporaryFile(mode='w', newline='', encoding='shift_jis', delete=False)
                fname = fp.name
                header=[]
                for key in content:
                    header.append(key)
                writer = csv.DictWriter(fp, fieldnames=header)
                writer.writeheader()
                writer.writerow(content)
                fp.close()

                fp = open(fname, 'r')

                zip_io = io.BytesIO()
                with zipfile.ZipFile(zip_io, mode='w', compression=zipfile.ZIP_DEFLATED) as backup_zip:
                    backup_zip.writestr('CSVファイル.csv', fp.read())
                response = HttpResponse(zip_io.getvalue(), content_type='application/x-zip-compressed')
                response['Content-Disposition'] = 'attachment; filename=%s' % 'your_zipfilename' + ".zip"
                response['Content-Length'] = zip_io.tell()
                fp.close()
                os.unlink(fname)
                return response

            else:
                # confirm
                context['method'] = 'confirm'
                context['title'] = []
                f_items = form_item['declared_fields']
                for item in f_items:
                    if type(content[item]) == type([]):
                        context['title'].append({'title': f_items[item].label, 'description': f_items[item].label_suffix, 'answer': ','.join(content[item])})
                    else:
                        context['title'].append({'title': f_items[item].label, 'description': f_items[item].label_suffix, 'answer': content[item]})

    # draw enquete form
    DynamicForm = DynamicEnqForm(content)
    context['enq_form'] = DynamicForm
    return render(request, "dynamic.html", context)

dynamic.html

<form action="" method="POST">
  {% csrf_token %}
  {% if method == 'confirm' %}
  <div>
    <div>
      {% for item in title %}
      <p>{{ item.title }}</p>
      <p>{{ item.description}}</p>
      <p>{{ item.answer}}</p>
      {% endfor %}
      {% for field in enq_form %}
      {{ field.as_hidden }}
      {% endfor %}
      <input type="hidden" name="method" value="regist" />
    </div>
  </div>
  {% else %}
  {% for field in enq_form.visible_fields %}
  <div>
    <div>
      <label for="{{ field.id_for_label }}">{{ field.label }}</label>
      <p>{{ field.field.label_suffix }}</p>
      {{ field }}
    </div>
  </div>
  {% endfor %}
  {% endif %}
  <input type="submit">
</form>

今回の主役はここのコード

views.py

            for key in formset.cleaned_data:
                if key != 'csrfmiddlewaretoken' and key != 'method':
                    content[key] = formset.cleaned_data[key]
            if 'method' in request.POST and request.POST['method'] == 'regist':
                # regist

                fp = tempfile.NamedTemporaryFile(mode='w', newline='', encoding='shift_jis', delete=False)
                fname = fp.name
                header=[]
                for key in content:
                    header.append(key)
                writer = csv.DictWriter(fp, fieldnames=header)
                writer.writeheader()
                writer.writerow(content)
                fp.close()

                fp = open(fname, 'r')

                zip_io = io.BytesIO()
                with zipfile.ZipFile(zip_io, mode='w', compression=zipfile.ZIP_DEFLATED) as backup_zip:
                    backup_zip.writestr('CSVファイル.csv', fp.read())
                response = HttpResponse(zip_io.getvalue(), content_type='application/x-zip-compressed')
                response['Content-Disposition'] = 'attachment; filename=%s' % 'your_zipfilename' + ".zip"
                response['Content-Length'] = zip_io.tell()
                fp.close()
                os.unlink(fname)
                return response

実際に試してみる(成功パターン)

クライアントはWindows10でブラウザはChromeを使ってます。
まずは成功パターンである無難な文字列の入力
デスクトップが汚かったのでc:\TEMPに保存しましたが、Excelが真っ白だったのでデスクトップに保存するようにしたら問題なく表示されました。

実際に試してみる(失敗パターン)

クライアントはWindows10でブラウザはChromeを使ってます。
「～」をフォームに入力します。

とまぁこんな感じで怒られます"(-""-)"

原因は…

Windowsの文字コードはShift-JISではなくShift-JISの亜種であるCP932
「～」はPython上でUnicodeに変換されてますがそのコードはU+FF5E
Windows上のIMEパッドで確認すると…Shift-JIS(CP932)では0x8160

Unicodeには、「波型」と「全角ティルダ」という二つの異なる「〜」が存在します。一般的には「波型」であるu'\u301c'が使われます。Shift_JISの「〜」'\x81\x60'はUnicodeの「波型」にマップされています。

詳しくはこちらの記事を参考にしてください。
Unicode波型問題 CP932とShift_JISは同じエンコーディングではない

文字化け対策コード

文字化けを防ぐため、
- クライアントから送信されてくる値についてはCP932でエンコードし、Shift_JISでデコードする。
- ファイルはShift_JISで作成
- ZIPに読み込ませる際はバイナリーモードで読み込む
を対策し無事にCSVが文字化けせずに出力されました。

views.py

from django.shortcuts import render
from .forms import EnqForm
from django import forms
from django.http import HttpResponse
import io
import os
import csv, zipfile
import tempfile

# Create your views here.
def dynamic(request):
    context = {}
    content = {}
    form_item = {}
    qs = []

    # define enquete fields
    qs.append( {'title':'title1', 'description': 'note1', 'type': 'text' , 'required': False} )
    qs.append( {'title':'title2', 'description': 'note2', 'type': 'text' , 'required': True} )
    qs.append( {'title':'title3', 'description': 'note3', 'type': 'radio' , 'required': False} )
    qs.append( {'title':'title4', 'description': 'note4', 'type': 'multi' , 'required': True} )
    qs.append( {'title':'title5', 'description': 'note5', 'type': 'radio' , 'required': True} )

    # create enquete form objects
    no = 0
    for q in qs:
        if q['type'] == 'text':
            form_item.update( { ('ans%d' % no): forms.CharField(label=q['title'], label_suffix=q['description'], required=q['required'], max_length=256) } )
        elif q['type'] == 'radio':
            form_item.update( { ('ans%d' % no): forms.ChoiceField(label=q['title'], label_suffix=q['description'], required=q['required'], choices=(('a','a'),('b','b'),('c','c'),('d','d')), widget=forms.RadioSelect) } )
        elif q['type'] == 'multi':
            form_item.update( { ('ans%d' % no): forms.MultipleChoiceField(label=q['title'], label_suffix=q['description'], required=q['required'], choices=(('a','a'),('b','b'),('c','c'),('d','d')), widget=forms.CheckboxSelectMultiple) } )
        no += 1

    # create dynamic form
    DynamicEnqForm = type('DynamicEnqForm', (EnqForm,), form_item )

    # get enquete answers
    if request.method == 'POST':
        formset = DynamicEnqForm(request.POST or None)
        if formset.is_valid():
            for key in formset.cleaned_data:
                if key != 'csrfmiddlewaretoken' and key != 'method':
                    if type(formset.cleaned_data[key]) != type([]):
                        content[key] = formset.cleaned_data[key].encode('cp932').decode('shift_jis')
                    else:
                        items = formset.cleaned_data[key]
                        buf = []
                        for item in items:
                            buf.append( item.encode('cp932').decode('shift_jis') )
                        content[key] = buf

            if 'method' in request.POST and request.POST['method'] == 'regist':
                # regist

                fp = tempfile.NamedTemporaryFile(mode='w', newline='', encoding='shift_jis', delete=False)
                fname = fp.name
                header=[]
                for key in content:
                    header.append(key)
                writer = csv.DictWriter(fp, fieldnames=header)
                writer.writeheader()
                writer.writerow(content)
                fp.close()

                fp = open(fname, 'rb' )

                zip_io = io.BytesIO()
                with zipfile.ZipFile(zip_io, mode='w', compression=zipfile.ZIP_DEFLATED) as backup_zip:
                    backup_zip.writestr('CSVファイル.csv', fp.read())
                response = HttpResponse(zip_io.getvalue(), content_type='application/x-zip-compressed')
                response['Content-Disposition'] = 'attachment; filename=%s' % 'your_zipfilename' + ".zip"
                response['Content-Length'] = zip_io.tell()
                fp.close()
                os.unlink(fname)
                return response

            else:
                # confirm
                context['method'] = 'confirm'
                context['title'] = []
                f_items = form_item['declared_fields']
                for item in f_items:
                    if type(content[item]) == type([]):
                        context['title'].append({'title': f_items[item].label, 'description': f_items[item].label_suffix, 'answer': ','.join(content[item])})
                    else:
                        context['title'].append({'title': f_items[item].label, 'description': f_items[item].label_suffix, 'answer': content[item]})

    # draw enquete form
    DynamicForm = DynamicEnqForm(content)
    context['enq_form'] = DynamicForm
    return render(request, "dynamic.html", context)

解説

まずはクライアントから送信された値を綺麗にShift_JISに変換します。
一度CP932でエンコードした後に改めてShift_JISに変換します。
ちょっと処理が複雑になっているのはlist型の場合もあるのでその処理が入ってます。

            for key in formset.cleaned_data:
                if key != 'csrfmiddlewaretoken' and key != 'method':
                    if type(formset.cleaned_data[key]) != type([]):
                        content[key] = formset.cleaned_data[key].encode('cp932').decode('shift_jis')
                    else:
                        items = formset.cleaned_data[key]
                        buf = []
                        for item in items:
                            buf.append( item.encode('cp932').decode('shift_jis') )
                        content[key] = buf

ファイルの出力部分は変わっていません。Shift_JISとして書きだします。

                fp = tempfile.NamedTemporaryFile(mode='w', newline='', encoding='shift_jis', delete=False)
                fname = fp.name
                header=[]
                for key in content:
                    header.append(key)
                writer = csv.DictWriter(fp, fieldnames=header)
                writer.writeheader()
                writer.writerow(content)
                fp.close()

あとはShift_JISで出力したファイルを読み込む際に、バイナリーモードで読み込みます。
そうしないとまたUnicodeになっちゃうので。

                fp = open(fname, 'rb' )