Python 批次處理資料轉換與自動化技術

import pandas as pd

# 載入 CSV 檔案
df = pd.read_csv('data.csv')

# 資料清理：移除缺失值
df = df.dropna()

# 資料轉換：將特定欄位轉換為數值型別
df['numeric_column'] = pd.to_numeric(df['numeric_column'], errors='coerce')

# 資料分組與彙總
df_grouped = df.groupby('category_column').sum()

# 輸出結果到新的 CSV 檔案
df_grouped.to_csv('processed_data.csv')

import dask.dataframe as dd

# 使用 Dask 載入大型資料集
ddf = dd.read_csv('large_data.csv')

# 進行與 Pandas 類別似的操作
ddf['new_column'] = ddf['existing_column'] * 2
ddf_grouped = ddf.groupby('category_column').sum()

# 計算結果
final_df = ddf_grouped.compute()

import sqlalchemy

# 建立資料函式庫連線引擎
engine = sqlalchemy.create_engine('sqlite:///processed_data.db')

# 將 DataFrame 輸出到 SQL 資料函式庫
df.to_sql('transformed_table', engine, index=False, if_exists='replace')

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta

def process_data_task():
    # 定義資料處理邏輯
    pass

dag = DAG('data_processing_dag', default_args={'owner': 'airflow', 'start_date': datetime(2023, 1, 1)}, schedule_interval='@daily')

process_data = PythonOperator(
    task_id='process_data',
    python_callable=process_data_task,
    dag=dag
)

import smtplib
from email.mime.text import MIMEText

# 電子郵件組態
smtp_server = 'smtp.example.com'
smtp_port = 587
sender_email = '[email protected]'
recipient_email = '[email protected]'
password = 'your_password'

# 建立電子郵件內容
subject = '自動化電子郵件'
body = '這是一封由 Python 指令碼傳送的自動化電子郵件。'
msg = MIMEText(body)
msg['Subject'] = subject
msg['From'] = sender_email
msg['To'] = recipient_email

try:
    with smtplib.SMTP(smtp_server, smtp_port) as server:
        server.starttls()  # 啟用 TLS 安全連線
        server.login(sender_email, password)
        server.send_message(msg)
    print("電子郵件傳送成功。")
except Exception as e:
    print(f"電子郵件傳送失敗：{e}")

from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication
from email.mime.text import MIMEText

# 建立多部分電子郵件
msg = MIMEMultipart()
msg['Subject'] = '自動化電子郵件（附附件）'
msg['From'] = sender_email
msg['To'] = recipient_email

# 新增郵件正文
body = MIMEText('請查閱附件中的檔案。')
msg.attach(body)

# 新增 PDF 附件
filename = 'report.pdf'
with open(filename, 'rb') as file:
    part = MIMEApplication(file.read(), Name=filename)
    part['Content-Disposition'] = f'attachment; filename="{filename}"'
    msg.attach(part)

# 使用之前的 SMTP 邏輯傳送電子郵件

import imaplib
import email

# IMAP 伺服器組態
imap_server = 'imap.example.com'
username = '[email protected]'
password = 'your_password'

try:
    # 連線到 IMAP 伺服器
    with imaplib.IMAP4_SSL(imap_server) as mail:
        mail.login(username, password)
        mail.select('inbox')  # 選擇收件箱
        
        # 搜尋所有電子郵件
        status, data = mail.search(None, 'ALL')
        mail_ids = data[0].split()
        
        for mail_id in mail_ids:
            # 根據 ID 取得電子郵件訊息
            status, msg_data = mail.fetch(mail_id, '(RFC822)')
            raw_email = msg_data[0][1]
            email_message = email.message_from_bytes(raw_email)
            
            # 提取電子郵件主題
            subject = email_message['subject']
            print(f'主題：{subject}')
except Exception as e:
    print(f"無法檢索電子郵件：{e}")

from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build

# 設定 OAuth 2.0 流程
flow = InstalledAppFlow.from_client_secrets_file(
    'client_secrets.json',
    scopes=['https://mail.google.com/']
)

# 認證並取得憑證
credentials = flow.run_local_server(port=0)

# 使用憑證與 Gmail API 互動
service = build('gmail', 'v1', credentials=credentials)

# 取得收件箱中的電子郵件
results = service.users().messages().list(userId='me', labelIds=['INBOX']).execute()
messages = results.get('messages', [])

for message in messages:
    msg = service.users().messages().get(userId='me', id=message['id']).execute()
    print(f"訊息片段：{msg['snippet']}")

from apscheduler.schedulers.blocking import BlockingScheduler

def send_scheduled_email():
    # 定義傳送電子郵件的邏輯
    print("已傳送排程電子郵件。")

scheduler = BlockingScheduler()
scheduler.add_job(send_scheduled_email, 'interval', hours=12)

try:
    scheduler.start()
except (KeyboardInterrupt, SystemExit):
    pass

import unittest

def add(a, b):
    return a + b

class TestMathOperations(unittest.TestCase):
    def test_add(self):
        self.assertEqual(add(2, 3), 5)
        self.assertEqual(add(-1, 1), 0)
        self.assertEqual(add(-1, -1), -2)

if __name__ == '__main__':
    unittest.main()

import pytest

def add(a, b):
    return a + b

def test_add():
    assert add(2, 3) == 5

from unittest import mock, TestCase

def get_quote():
    # 假設這裡呼叫了一個外部 API
    pass

class TestQuoteFunction(TestCase):
    @mock.patch('__main__.get_quote', return_value="Mock Quote")
    def test_get_quote(self, mock_get_quote):
        result = get_quote()
        self.assertEqual(result, "Mock Quote")

Feature: Addition
  Scenario: Add two numbers
    Given the user inputs 2 and 3
    When the system calculates the sum
    Then the output should be 5

from locust import HttpUser, TaskSet, task

class UserBehavior(TaskSet):
    @task(1)
    def index(self):
        self.client.get("/")

    @task(2)
    def about(self):
        self.client.get("/about")

class WebsiteUser(HttpUser):
    tasks = [UserBehavior]
    min_wait = 5000
    max_wait = 9000

name: Python application test
on: [push]
jobs:
  build:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
      - name: Set up Python
        uses: actions/setup-python@v2
        with:
          python-version: '3.x'
      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install pytest
      - name: Test with pytest
        run: |
          pytest

# 安裝 coverage.py
pip install coverage

# 執行 coverage.py
coverage run -m pytest
coverage report