使用Pandas与CSV库处理CSV时的后台任务及更新报错问题

我的场景与问题 #

我正在开发一个简易应用，希望每15分钟在后台更新CSV文件中的部分值，避免阻塞应用界面，但没能达到预期效果。相关代码和遇到的问题如下：

用到的依赖 #

我用到了pandas、sched和time这些库，还有Flask的路由。

代码片段 #

# INTERFACE
@app.route("/")
def home():
    s = sched.scheduler(time.time, time.sleep)
    s.enter(15, 1, timer)
    s.run()
    return render_template("index.html")

# TIMER
def timer():
    timer = 0
    csv_file='C:\\Python27\\Walmart\\sheet.csv'
    print("UPDATING")
    data_df = pd.read_csv(csv_file)
    print("READ")
    for i, row in data_df.iterrows() :
        sku = data_df.iloc[i]['Walmart SKU']
        print (sku)
        if sku is '':
            break
        else:
            update(sku)
            print("Item Updated")
    print("UPDATECOMPLETE")
    home()

# UPDATE
def update(sku):
    lookup=str(sku)
    lookup = lookup.replace('.0', '')
    product = wapy.product_lookup(lookup)
    ts = time.time()
    st = datetime.datetime.today().strftime('%Y-%m-%d %I:%M %p')
    print (product.name)
    if product.available_online is 'TRUE':
        instock = 'yes'
    else:
        instock ='no'
    quote_page = product.product_url
    page = urlopen(quote_page)
    soup = BeautifulSoup(page, 'html.parser')
    sold_box = soup.find('a', attrs={'class': 'font-bold prod-SoldShipByMsg'})
    sold = sold_box.text.strip()
    left_box = soup.find('div', attrs={'class': 'prod-ProductOffer-urgencyMsg'})
    left = left_box.text.strip()
    if left is '':
        stock=product.stock
    else:
        stock=left
    # fields=[lookup + ',' + '$'+str(product.sale_price) + ',' + instock + ',' + stock + ',' + str(sold) + ',' + st + ',' + '$'+str(product.msrp)]
    pathto_csv = 'C:\\Python27\\Walmart\\sheet.csv'
    data_df = pd.read_csv(pathto_csv)
    print("CSV READ")
    data_df.set_value([lookup], ['Price'], '$'+str(product.sale_price))
    data_df.set_value([lookup], ['In Stock'], instock )
    data_df.set_value([lookup], ['Quantity'], stock)
    data_df.set_value([lookup], ['Last Update'], str(sold))
    data_df.set_value([lookup], ['Min Price'], '$'+str(product.msrp))
    data_df.to_csv(pathto_csv)
    with open(r'sheet.csv', 'a') as f:
        writer = csv.writer(f, delimiter=' ', quotechar = ' ')
        writer.writerow(fields)
    print(st)
    print("UPDATED! 15 Minutes Have Passed!")

遇到的两个问题 #

1. KeyError报错：执行更新时出现KeyError: "['879091509'] not in index"，我原本以为用data_df.set_value([lookup], ['Price'], 值)可以通过SKU（也就是lookup）定位行来更新对应单元格，但显然没成功。
2. 界面阻塞：完全没法访问应用界面，推测是定时器重置时重复执行代码，把界面给卡住了。

现在需要解决的核心问题：怎么让CSV更新任务只在后台运行，不影响前端界面，同时解决KeyError的问题？

解决方案：分两步解决你的问题 #

一、先搞定KeyError：正确用SKU定位行更新 #

你遇到的KeyError是因为set_value（这个方法在pandas新版本里已经被loc/iloc取代了）默认按行索引查找，但你的CSV行索引不是SKU，lookup是Walmart SKU列的取值，不是DataFrame的索引。

修正方法很简单：

1. 读取CSV时，把Walmart SKU列设为DataFrame的索引，同时指定该列为字符串类型（避免数字自动转成带.0的格式）；
2. 改用loc方法更新（比过时的set_value更稳定）。

修改update函数里的读取和更新逻辑：

pathto_csv = 'C:\\Python27\\Walmart\\sheet.csv'
# 读取时指定Walmart SKU为索引，确保列类型是字符串
data_df = pd.read_csv(pathto_csv, index_col='Walmart SKU', dtype={'Walmart SKU': str})
print("CSV READ")
# 用loc直接通过SKU（索引）定位行和列更新
data_df.loc[lookup, 'Price'] = '$' + str(product.sale_price)
data_df.loc[lookup, 'In Stock'] = instock
data_df.loc[lookup, 'Quantity'] = stock
data_df.loc[lookup, 'Last Update'] = str(sold)
data_df.loc[lookup, 'Min Price'] = '$' + str(product.msrp)
# 保存时不要把索引写入成新列，设置index=False
data_df.to_csv(pathto_csv, index=False)

另外，timer函数里的data_df.iloc[i]['Walmart SKU']可以简化成row['Walmart SKU']，因为iterrows()返回的每一行本身就是Series，直接取列名即可。

二、解决界面阻塞：让更新任务在后台线程跑 #

你现在的问题是home路由里直接调用了s.run()，这会阻塞Flask的请求处理线程——sched是同步执行的，加上timer最后递归调用home()，相当于一直在循环执行定时器，根本没机会返回页面给前端。

正确的做法是用后台线程单独执行定时更新任务，路由只负责返回页面：

from threading import Thread
import time

# 后台定时更新函数
def background_update_task():
    while True:
        print("UPDATING")
        csv_file='C:\\Python27\\Walmart\\sheet.csv'
        data_df = pd.read_csv(csv_file, dtype={'Walmart SKU': str})
        print("READ")
        for _, row in data_df.iterrows():
            sku = row['Walmart SKU']
            print(sku)
            if not sku:  # 空字符串判断用not更稳妥，不要用is ''
                break
            update(sku)
            print("Item Updated")
        print("UPDATECOMPLETE")
        # 等待15分钟（注意你之前写的15是秒！这里改成15*60秒）
        time.sleep(15 * 60)

# INTERFACE
@app.route("/")
def home():
    # 路由只做一件事：返回页面
    return render_template("index.html")

# 应用启动时启动后台线程
if __name__ == '__main__':
    # daemon=True表示主线程退出时，后台线程也跟着退出
    Thread(target=background_update_task, daemon=True).start()
    app.run(debug=True)

这样Flask主线程专门处理前端请求，后台线程单独跑定时更新，就不会阻塞界面了。

额外小细节提醒 #

• 你之前的sched设置的是15秒，不是15分钟，记得改成15*60；
• 空字符串判断不要用is ''，应该用if not sku或者sku == ''，is是判断对象身份，可能出现意外问题；
• update函数里的fields变量被注释了，但后面还在用writer.writerow(fields)，会报错，记得要么取消注释要么删除这段代码。

内容的提问来源于stack exchange，提问作者Anne