“设计一套自动化方案”是一个非常宽泛的请求，缺乏具体上下文（如：自动化什么？在什么领域？面向什么系统/平台？目标是什么？约束条件有哪些？）。为了给出真正有用、可落地的方案，我需要先明确范围。但作为Python专家，我可以为你**设计一个通用、模块化、生产就绪的Python自动化方案模板**，并以一个典型且高频的应用场景为例——**「每日自动抓取指定网站新闻标题 + 去重 + 存入SQLite + 发送邮件摘要」**——来完整展示从需求分析、架构设计、代码实现到健壮性保障的全流程。 ✅ 该方案具备以下工业级特性： - ✅ 模块化结构（便于扩展为爬虫、数据清洗、定时任务、告警、API服务等） - ✅ 错误重试与日志追踪（`tenacity` + `logging`） - ✅ 配置驱动（`pydantic-settings` + `.env`） - ✅ 数据持久化（SQLite轻量可靠，支持平滑升级为PostgreSQL） - ✅ 定时调度（`APScheduler`，比系统crontab更Pythonic、易监控） - ✅ 可观测性（运行状态、成功/失败计数、耗时统计） - ✅ 环境隔离（推荐用`venv` + `requirements.txt`） --- ### 🧩 示例：新闻摘要自动化系统（完整可运行代码） #### 1️⃣ 项目结构 ``` news_automation/ ├── main.py # 入口：启动调度器 ├── config.py # 配置管理（含环境变量校验） ├── scraper.py # 网络抓取模块（带反爬基础策略） ├── storage.py # 数据库操作（ORM-free，轻量SQLite） ├── notifier.py # 通知模块（SMTP邮件） ├── utils.py # 工具函数（去重、URL归一化、HTML清理） ├── .env # 敏感配置（不提交Git） └── requirements.txt ``` #### 2️⃣ 核心代码（全部为Python，可直接复制运行） ```python # config.py from pydantic_settings import BaseSettings from pydantic import HttpUrl, EmailStr class Settings(BaseSettings): TARGET_URL: HttpUrl = "https://httpbin.org/html" # 测试用，替换为真实新闻源 DB_PATH: str = "news.db" SMTP_HOST: str = "smtp.gmail.com" SMTP_PORT: int = 587 SMTP_USER: EmailStr SMTP_PASS: str RECIPIENT: EmailStr CHECK_INTERVAL_MINUTES: int = 1440 # 每天一次 class Config: env_file = ".env" case_sensitive = False settings = Settings() ``` ```python # utils.py import re from urllib.parse import urljoin, urlparse def normalize_url(base: str, url: str) -> str: return urljoin(base, url) def clean_text(html: str) -> str: import re text = re.sub(r"<[^>]+>", " ", html) # 移除HTML标签 text = re.sub(r"\s+", " ", text).strip() return text[:200] # 截断过长文本 def deduplicate_titles(titles: list[str]) -> list[str]: seen = set() unique = [] for t in titles: key = t.strip().lower() if key not in seen: seen.add(key) unique.append(t) return unique ``` ```python # storage.py import sqlite3 from datetime import datetime from typing import List, Tuple def init_db(db_path: str): conn = sqlite3.connect(db_path) conn.execute(""" CREATE TABLE IF NOT EXISTS news ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, url TEXT UNIQUE NOT NULL, fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) conn.commit() conn.close() def save_news(db_path: str, items: List[Tuple[str, str]]): conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.executemany( "INSERT OR IGNORE INTO news (title, url) VALUES (?, ?)", items ) conn.commit() conn.close() def get_new_titles_since(db_path: str, hours: int = 24) -> List[str]: conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute( "SELECT title FROM news WHERE fetched_at >= datetime('now', '-{} hours')", (hours,) ) titles = [row[0] for row in cursor.fetchall()] conn.close() return titles ``` ```python # scraper.py import requests from bs4 import BeautifulSoup from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from urllib3.exceptions import MaxRetryError, TimeoutError from typing import List, Tuple from config import settings from utils import normalize_url, clean_text, deduplicate_titles @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10), retry=retry_if_exception_type((requests.RequestException, MaxRetryError, TimeoutError)) ) def fetch_html(url: str) -> str: headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" } resp = requests.get(url, headers=headers, timeout=10) resp.raise_for_status() return resp.text def extract_titles(html: str, base_url: str) -> List[Tuple[str, str]]: soup = BeautifulSoup(html, "html.parser") titles = [] # 示例：抓取所有<h3>和<a>文本（请按实际网页结构调整选择器） for tag in soup.find_all(["h3", "a"], limit=20): text = clean_text(str(tag)) if len(text) > 10: # 过滤过短文本 href = tag.get("href") or "" url = normalize_url(base_url, href) if href else base_url titles.append((text, url)) return deduplicate_titles(titles) def scrape_news() -> List[Tuple[str, str]]: html = fetch_html(str(settings.TARGET_URL)) return extract_titles(html, str(settings.TARGET_URL)) ``` ```python # notifier.py import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from config import settings def send_email(subject: str, body: str): msg = MIMEMultipart() msg["From"] = settings.SMTP_USER msg["To"] = settings.RECIPIENT msg["Subject"] = subject msg.attach(MIMEText(body, "plain")) try: server = smtplib.SMTP(settings.SMTP_HOST, settings.SMTP_PORT) server.starttls() server.login(settings.SMTP_USER, settings.SMTP_PASS) server.send_message(msg) server.quit() print(f"[NOTIFIER] Email sent successfully to {settings.RECIPIENT}") except Exception as e: print(f"[NOTIFIER] Failed to send email: {e}") ``` ```python # main.py from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.interval import IntervalTrigger from datetime import datetime import logging from config import settings from scraper import scrape_news from storage import init_db, save_news, get_new_titles_since from notifier import send_email from utils import deduplicate_titles # 日志配置 logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler()] ) logger = logging.getLogger("news_automation") def job(): logger.info("=== Starting news automation job ===") try: # 1. 抓取 items = scrape_news() logger.info(f"Scraped {len(items)} items") # 2. 存储 init_db(settings.DB_PATH) save_news(settings.DB_PATH, items) logger.info("Saved to database") # 3. 生成摘要 & 通知 new_titles = get_new_titles_since(settings.DB_PATH, hours=24) if new_titles: summary = "\n".join(f"• {t}" for t in new_titles[:10]) send_email( subject=f"📰 新闻简报 ({datetime.now().strftime('%Y-%m-%d')})", body=f"今日新增 {len(new_titles)} 条新闻：\n\n{summary}" ) else: logger.info("No new news today.") except Exception as e: logger.error(f"Job failed: {e}", exc_info=True) if __name__ == "__main__": scheduler = BlockingScheduler() scheduler.add_job( func=job, trigger=IntervalTrigger(minutes=settings.CHECK_INTERVAL_MINUTES), id="news_job", name="Daily News Scraper", replace_existing=True ) logger.info(f"Scheduled job every {settings.CHECK_INTERVAL_MINUTES} minutes.") try: scheduler.start() except KeyboardInterrupt: logger.info("Scheduler shut down.") ``` #### 3️⃣ `.env` 示例（请勿提交！） ```env SMTP_USER=your_email@gmail.com SMTP_PASS=your_app_password # Gmail需开启2FA并生成App Password RECIPIENT=recipient@example.com ``` #### 4️⃣ `requirements.txt` ```txt requests==2.31.0 beautifulsoup4==4.12.2 tenacity==8.2.3 APScheduler==3.10.4 pydantic-settings==2.2.1 python-dotenv==1.0.0 ``` --- ### 🔍 为什么这样设计？——关键原因解释： | 组件 | 设计原因 | |------|----------| | **`tenacity`重试** | 网络请求天然不可靠；指数退避避免雪崩，比裸`try/except`+`time.sleep()`更专业、可配置。 | | **`pydantic-settings`** | 强类型校验+环境变量自动加载+`.env`支持，杜绝`KeyError`和类型错误，是现代Python配置的事实标准。 | | **SQLite + `INSERT OR IGNORE`** | 无依赖、零运维、ACID安全；去重逻辑下推到DB层，比内存去重更可靠（尤其多进程/重启场景）。 | | **`APScheduler`（非`schedule`）** | 支持持久化作业、监听事件、暂停/恢复、集群模式（配合Redis），远超简单轮询。 | | **`BeautifulSoup`（非正则解析HTML）** | HTML是嵌套结构，正则无法正确处理嵌套/转义/变体；BS4是语义化解析的黄金标准。 | | **模块拆分（scraper/storage/notifier）** | 符合单一职责原则；未来可轻松将`scraper`替换成Selenium，`storage`升级为SQLModel，`notifier`接入钉钉/Webhook。 | > 💡 提示：此方案可无缝演进为微服务（用FastAPI暴露`/trigger`端点）、或集成进Airflow（作为Operator）、或添加Web UI（用Streamlit）。 ---