如何修复：“sqlite3.OperationalError：无法解码为 UTF-8”答案

【问题标题】：How to fix: "sqlite3.OperationalError: Could not decode to UTF-8"如何修复：“sqlite3.OperationalError：无法解码为 UTF-8”
【发布时间】：2019-10-29 00:25:30
【问题描述】：

使用 Python sqlite3 包，我从 sqlite 数据库（带有三重内连接）中选择了一些行，以便稍后将它们放入 pandas 数据框中。

我尝试运行以下代码从数据库中进行选择：

import pandas as pd
import sqlite3

# connect to the database
conn = sqlite3.connect("C:\Quant\Data_Storage2\sqlite\db\scorecard.db")
conn.text_factory = str
c = conn.cursor()

# create pandas df with data from the sqlite db
sql = '''SELECT MACRO_SA.SERIES_ID, MACRO_SA.COUNTRY_ID, MACRO_SA.DATA, 
         MACRO_SA.REL_DATE, HF_DATA.DATA
         FROM MACRO_SA
            INNER JOIN MAP_SERIES ON MAP_SERIES.CUR = MACRO_SA.COUNTRY_ID
            INNER JOIN HF_DATA ON MAP_SERIES.CUR = MACRO_SA.COUNTRY_ID
         WHERE (MAP_SERIES.CRNCY NOT NULL)  
            AND (HF_DATA.SERIES_ID = 'bg_cur')
'''
df_to_adj = c.execute(sql).fetchall()
conn.commit()
df_to_adj = pd.DataFrame(df_to_adj, columns=['SERIES_ID', 'COUNTRY_ID', 'DATA', 'REL_DATE', 'FX_RATE'])

我收到以下错误：“sqlite3.OperationalError：无法解码为 UTF-8”。非常感谢任何帮助。

【问题讨论】：

标签： python database sqlite

【解决方案1】：

conn.text_factory = str

适用于 Python 2；如果您使用的是 Python 3，则必须提供自己的魔术函数，该函数以字节为单位，并吐出您想要的字符串。一种选择是尝试解码字符串，如果失败，则返回原始字节。如果你总是可以处理字节，那就做

conn.text_factory = bytes

就够了。有关血腥细节，请参阅https://docs.python.org/3.7/library/sqlite3.html。

【讨论】：

我遇到了类似的问题。有问题的列是二进制数据的 blob，因此将其解释为字节应该可以工作，但是当我将连接 text_factory 设置为字节时，它仍然给我相同的错误消息。我通常使用会话，但即使我从引擎获取连接，修改该连接并在该连接上使用执行，它仍然给我这个错误。

【解决方案2】：

我遇到了类似的问题。数据库中的文本似乎以另一种编码而不是 utf-8 编码。为了找出哪些编码有效，我使用了一个辅助函数来尝试所有编码：

from encodings.aliases import aliases
_encodings = set(aliases.values())

def try_encodings(byte_text: bytes):
    for encoding in _encodings:
        try:
            print(f'Encoding {encoding}: {byte_text.decode(encoding)}')
        except (UnicodeDecodeError, LookupError):
            pass

将text_factory 设置为bytes 并逐行迭代并解码问题列。如果发现有解码问题的行，尝试所有其他编码：

c = conn.cursor()
conn.text_factory = bytes
sql = 'Select * from ...'
c.execute(sql)

col_ix = 0  # Set column index of problem column
while row := c.fetchone():
    try:
        text = row[col_ix].decode()
    except UnicodeDecodeError as e:
        print('Cannot decode byte string: ', row[col_ix])
        try_encodings(row[col_ix])
        break

如果您找到正确的编码，您可以将其设置为 text_factory。比如cp852：

conn.text_factory = lambda x: x.decode('cp852')

【讨论】：