conda install openpyxl pandas xlrd xlwt xlsxwriter

from openpyxl import load_workbook

# 加載工作簿
wb = load_workbook('test_data.xlsx')

# 獲取工作表
sheet = wb['Sheet1']

# 讀取單元格值
print(sheet['A1'].value)  # 讀取A1單元格
print(sheet.cell(row=2, column=1).value)  # 讀取第2行第1列

# 遍歷數(shù)據(jù)
for row in sheet.iter_rows(min_row=2, values_only=True):
    print(row)  # 輸出每行數(shù)據(jù)（跳過標(biāo)題行）

import pandas as pd

# 讀取整個工作簿
all_sheets = pd.read_excel('test_data.xlsx', sheet_name=None)
sales_data = all_sheets['Sheet1']  # 獲取銷售數(shù)據(jù)表

# 讀取指定工作表
customer_data = pd.read_excel('test_data.xlsx', sheet_name='Sheet2')

# 顯示前5行
print(sales_data.head())

# 處理缺失值
sales_data.fillna(0, inplace=True)  # 填充缺失值為0
# 或刪除缺失行
sales_data.dropna(inplace=True)

# 數(shù)據(jù)類型轉(zhuǎn)換
sales_data['單價'] = sales_data['單價'].astype(float)
sales_data['日期'] = pd.to_datetime(sales_data['日期'])

# 字符串處理
sales_data['產(chǎn)品'] = sales_data['產(chǎn)品'].str.strip()  # 去除空格

# 刪除重復(fù)值
sales_data.drop_duplicates(subset=['訂單號'], inplace=True)

# 計算銷售總額
total_sales = (sales_data['數(shù)量'] * sales_data['單價']).sum()
print(f"總銷售額: {total_sales:.2f}")

# 按產(chǎn)品分組統(tǒng)計
product_stats = sales_data.groupby('產(chǎn)品').agg({
    '數(shù)量': 'sum',
    '單價': 'mean'
}).reset_index()

# 篩選高價值客戶
high_value_customers = sales_data.groupby('客戶ID')['金額'].sum().nlargest(10)

# 創(chuàng)建新DataFrame
result = pd.DataFrame({
    '產(chǎn)品': ['A', 'B', 'C'],
    '總銷量': [1200, 850, 630],
    '平均單價': [25.5, 32.8, 19.9]
})

# 寫入新文件
result.to_excel('sales_summary.xlsx', index=False)

# 寫入多個工作表
with pd.ExcelWriter('multi_sheet.xlsx') as writer:
    product_stats.to_excel(writer, sheet_name='產(chǎn)品統(tǒng)計', index=False)
    high_value_customers.to_excel(writer, sheet_name='高價值客戶')

from openpyxl.styles import Font, Alignment, PatternFill
from openpyxl.utils.dataframe import dataframe_to_rows

# 創(chuàng)建新工作簿
wb = Workbook()
ws = wb.active
ws.title = "銷售匯總"

# 寫入數(shù)據(jù)（帶格式）
for r_idx, row in enumerate(dataframe_to_rows(result, index=False, header=True), 1):
    for c_idx, value in enumerate(row, 1):
        cell = ws.cell(row=r_idx, column=c_idx, value=value)
        # 設(shè)置標(biāo)題行格式
        if r_idx == 1:
            cell.font = Font(bold=True)
            cell.alignment = Alignment(horizontal='center')
        # 設(shè)置金額列格式
        if c_idx == 3 and r_idx > 1:
            cell.number_format = '#,##0.00'

# 設(shè)置列寬
ws.column_dimensions['A'].width = 15
ws.column_dimensions['B'].width = 10

# 保存文件
wb.save('formatted_report.xlsx')

import pandas as pd
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders
import os

def process_sales_data():
    # 1. 數(shù)據(jù)合并
    all_data = pd.DataFrame()
    for store in ['store1', 'store2', 'store3']:  # 實際應(yīng)遍歷所有分店
        df = pd.read_excel(f'{store}_data.xlsx')
        df['分店'] = store
        all_data = pd.concat([all_data, df])
    
    # 2. 數(shù)據(jù)分析
    # 產(chǎn)品銷量排名
    product_rank = all_data.groupby('產(chǎn)品')['數(shù)量'].sum().sort_values(ascending=False).head(10)
    
    # 區(qū)域銷售對比
    region_sales = all_data.groupby('地區(qū)')['金額'].sum()
    
    # 庫存預(yù)警（假設(shè)庫存數(shù)據(jù)在另一個文件）
    inventory = pd.read_excel('inventory.xlsx')
    alert_items = inventory[inventory['庫存量'] < inventory['預(yù)警值']]
    
    # 3. 生成報表
    wb = Workbook()
    
    # 產(chǎn)品銷量表
    ws1 = wb.active
    ws1.title = "產(chǎn)品銷量排名"
    ws1.append(['排名', '產(chǎn)品', '總銷量'])
    for i, (product, qty) in enumerate(product_rank.items(), 1):
        ws1.append([i, product, qty])
    
    # 設(shè)置格式
    for row in ws1.iter_rows(min_row=1, max_row=1):
        for cell in row:
            cell.font = Font(bold=True)
            cell.fill = PatternFill(start_color='FFFF00', end_color='FFFF00', fill_type='solid')
    
    # 區(qū)域銷售表
    ws2 = wb.create_sheet("區(qū)域銷售對比")
    ws2.append(['地區(qū)', '銷售額'])
    for region, sales in region_sales.items():
        ws2.append([region, sales])
    
    # 庫存預(yù)警表
    ws3 = wb.create_sheet("庫存預(yù)警")
    if not alert_items.empty:
        for r_idx, row in enumerate(dataframe_to_rows(alert_items, index=False, header=True), 1):
            for c_idx, value in enumerate(row, 1):
                ws3.cell(row=r_idx, column=c_idx, value=value)
    else:
        ws3.append(["無庫存預(yù)警"])
    
    # 保存文件
    report_file = 'daily_sales_report.xlsx'
    wb.save(report_file)
    return report_file

def send_email(report_file):
    msg = MIMEMultipart()
    msg['From'] = 'report@example.com'
    msg['To'] = 'manager@example.com'
    msg['Subject'] = '每日銷售報表'
    
    # 添加附件
    with open(report_file, 'rb') as f:
        part = MIMEBase('application', 'octet-stream')
        part.set_payload(f.read())
        encoders.encode_base64(part)
        part.add_header('Content-Disposition', f'attachment; filename="{os.path.basename(report_file)}"')
        msg.attach(part)
    
    # 發(fā)送郵件（實際需要配置SMTP服務(wù)器）
    # with smtplib.SMTP('smtp.example.com') as server:
    #     server.login('username', 'password')
    #     server.send_message(msg)
    print("郵件發(fā)送模擬完成")

# 執(zhí)行流程
if __name__ == "__main__":
    report = process_sales_data()
    send_email(report)

chunk_size = 10000
chunks = pd.read_excel('large_file.xlsx', chunksize=chunk_size)
for chunk in chunks:
    process_chunk(chunk)  # 處理每個數(shù)據(jù)塊

import sqlite3
conn = sqlite3.connect(':memory:')  # 使用內(nèi)存數(shù)據(jù)庫
sales_data.to_sql('sales', conn, index=False)
# 然后在數(shù)據(jù)庫中進(jìn)行復(fù)雜查詢

dtype_dict = {
    '日期': 'datetime64[ns]',
    '產(chǎn)品': 'category',  # 分類類型節(jié)省內(nèi)存
    '數(shù)量': 'int32',
    '單價': 'float32'
}
data = pd.read_excel('data.xlsx', dtype=dtype_dict)

import gc
del large_df
gc.collect()

from multiprocessing import Pool

def process_file(file_path):
    df = pd.read_excel(file_path)
    # 處理邏輯...
    return result

if __name__ == '__main__':
    files = ['file1.xlsx', 'file2.xlsx', 'file3.xlsx']
    with Pool(processes=4) as pool:  # 使用4個進(jìn)程
        results = pool.map(process_file, files)

# 讀取公式
print(sheet['A1'].value)  # 顯示公式文本如"=SUM(B1:B10)"

# 寫入公式
from openpyxl.formula.translate import Translator
ws['C1'] = "=SUM(A1:B1)"

from openpyxl import load_workbook
from openpyxl.utils.dataframe import dataframe_to_rows

# 加載模板文件
template = load_workbook('template.xlsx')
ws = template.active

# 寫入數(shù)據(jù)（保留原有格式）
for r_idx, row in enumerate(dataframe_to_rows(df, index=False, header=False), 2):
    for c_idx, value in enumerate(row, 1):
        ws.cell(row=r_idx, column=c_idx, value=value)

template.save('output.xlsx')

from sqlalchemy import create_engine
import pandas as pd

# 數(shù)據(jù)庫連接
engine = create_engine('sqlite:///sales.db')

# Excel到數(shù)據(jù)庫
df = pd.read_excel('data.xlsx')
df.to_sql('sales_table', engine, if_exists='replace')

# 數(shù)據(jù)庫到Excel
query_result = pd.read_sql('SELECT * FROM sales_table WHERE date > "2023-01-01"', engine)
query_result.to_excel('filtered_data.xlsx', index=False)