用 Python 做了一个AI日报生成器，每天自动汇总行业动态

pip install requests beautifulsoup4 schedule openpyxl

import requests
from bs4 import BeautifulSoup
import json
import time
import os
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import datetime

# ============ 配置区域 ============
LLM_API_KEY = "sk-your-deepseek-api-key"
LLM_API_URL = "https://api.deepseek.com/chat/completions"

# 邮件发送配置（可选，不发邮件也行）
EMAIL_ENABLED = False
SMTP_SERVER = "smtp.qq.com"
SMTP_PORT = 465
EMAIL_FROM = "your-email@qq.com"
EMAIL_PASSWORD = "your-smtp-password"
EMAIL_TO = ["boss@example.com", "team@example.com"]

# 新闻源配置：网站名称和对应的爬取规则
NEWS_SOURCES = [
    {
        "name": "36氪",
        "url": "https://36kr.com/newsflashes",
        "parser": "parse_36kr"
    },
    {
        "name": "物流沙龙",
        "url": "https://www.logclub.com/news",
        "parser": "parse_generic"
    },
    {
        "name": "亿邦动力",
        "url": "https://www.ebrun.com/news/",
        "parser": "parse_generic"
    }
]

# 通用请求头，模拟浏览器
HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
                  "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8"
}


# ============ 第一步：新闻爬取 ============
def fetch_page(url, timeout=15):
    """
    请求网页并返回BeautifulSoup对象
    加入重试机制和错误处理
    """
    try:
        response = requests.get(url, headers=HEADERS, timeout=timeout)
        response.raise_for_status()
        # 自动检测编码
        response.encoding = response.apparent_encoding
        soup = BeautifulSoup(response.text, "html.parser")
        return soup
    except requests.RequestExceptionase:
        print(f"  [请求失败] {url}: {e}")
        return None


def parse_36kr(url):
    """
    解析36氪快讯页面
    提取快讯标题和简要内容
    """
    soup = fetch_page(url)
    if not soup:
        return []

    news_list = []
    # 36氪快讯的条目（页面结构可能变化，需根据实际调整选择器）
    items = soup.select(".newsflash-item") orsoup.select("[class*='flash']") orsoup.select("article")

    for item in items[:20]:  # 取前20条
        title_elem = item.select_one("a") or item.select_one("h3") or item.select_one(".title")
        if not title_elem:
            continue

        title = title_elem.get_text(strip=True)
        link = title_elem.get("href", "")
        if link and not link.startswith("http"):
            link = "https://36kr.com"+link

        # 尝试获取摘要
        desc_elem = item.select_one(".description") or item.select_one("p") or item.select_one(".content")
        desc = desc_elem.get_text(strip=True) if desc_elem else""

        news_list.append({
            "title": title,
            "description": desc[:200],  # 摘要截断到200字
            "url": link,
            "source": "36氪"
        })

    return news_list


def parse_generic(url):
    """
    通用的新闻列表解析器
    适用于大多数新闻网站的文章列表页
    """
    soup = fetch_page(url)
    if not soup:
        return []

    news_list = []

    # 尝试多种常见的文章列表选择器
    articles = (
        soup.select("article") or
        soup.select(".news-item") or
        soup.select(".article-item") or
        soup.select(".post-item") or
        soup.select("li.item") or
        soup.select(".list-item")
    )

    for article in articles[:20]:
        # 提取标题
        title_elem = (
            article.select_one("h2 a") or
            article.select_one("h3 a") or
            article.select_one(".title a") or
            article.select_one("a")
        )
        if not title_elem:
            continue

        title = title_elem.get_text(strip=True)
        if len(title) <5:  # 太短的不是有效标题
            continue

        link = title_elem.get("href", "")
        if link and not link.startswith("http"):
            # 拼接完整URL
            from urllib.parse import urlparse
            parsed = urlparse(url)
            base_url = f"{parsed.scheme}://{parsed.netloc}"
            if link.startswith("/"):
                link = base_url+link
            else:
                link = base_url+"/"+link

        # 提取摘要
        desc_elem = (
            article.select_one("p") or
            article.select_one(".summary") or
            article.select_one(".desc") or
            article.select_one(".excerpt")
        )
        desc = desc_elem.get_text(strip=True) if desc_elem else""

        # 从URL推断来源名称
        from urllib.parse import urlparse
        source_name = urlparse(url).netloc.replace("www.", "")

        news_list.append({
            "title": title,
            "description": desc[:200],
            "url": link,
            "source": source_name
        })

    return news_list


# 解析器注册表
PARSERS = {
    "parse_36kr": parse_36kr,
    "parse_generic": parse_generic
}


def crawl_all_sources():
    """
    爬取所有配置的新闻源
    """
    all_news = []

    for source in NEWS_SOURCES:
        name = source["name"]
        url = source["url"]
        parser_name = source["parser"]

        print(f"正在爬取：{name} ({url})")

        parser_func = PARSERS.get(parser_name)
        if not parser_func:
            print(f"  [跳过] 未找到解析器：{parser_name}")
            continue

        news = parser_func(url)
        print(f"  获取到 {len(news)} 条新闻")
        all_news.extend(news)

        # 爬取间隔，别太猛
        time.sleep(2)

    returnall_news


# ============ 第二步：AI智能摘要 ============
def generate_daily_report(news_list):
    """
    把爬取到的新闻交给大模型，生成结构化的行业日报

    参数:
        news_list: 新闻列表，每条包含title、description、source

    返回:
        Markdown格式的日报文本
    """
    ifnot news_list:
        return"# 今日行业简报\n\n暂未抓取到新闻，请检查新闻源配置。"

    # 把新闻列表整理成文本，方便大模型阅读
    news_text = ""
    for i, item in enumerate(news_list, 1):
        news_text += f"{i}. [{item['source']}] {item['title']}\n"
        if item['description']:
            news_text += f"   摘要：{item['description']}\n"
        news_text += "\n"

    # 构造提示词
    prompt = f"""你是一个资深的行业分析师，请根据以下新闻列表生成一份行业日报。

今天的日期是：{datetime.now().strftime('%Y年%m月%d日')}

新闻列表：
{news_text}

请生成一份Markdown格式的行业日报，要求：

1. 按主题分类整理（如：物流与供应链、SaaS与企业服务、融资与创投、政策法规等）
2. 每个分类下挑3-5条最重要的新闻，写一段50字左右的点评
3. 在开头写一段"今日要点"（3-4句话概括当天最值得关注的事）
4. 在结尾写一段"编辑观点"（从行业趋势角度给出你的观察，3-5句话）
5. 忽略广告、软文和重复新闻
6. 语言风格：专业但不晦涩，像一个懂行的朋友在跟你聊今天发生了什么

请直接输出Markdown内容，不要加代码块标记。"""

    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {LLM_API_KEY}"
    }

    payload = {
        "model": "deepseek-chat",
        "messages": [
            {
                "role": "system",
                "content": "你是一个专业的行业分析师，擅长从大量新闻中提炼关键信息，输出结构清晰、观点鲜明的简报。"
            },
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.4,  # 适中温度，既不太死板也不太放飞
        "max_tokens": 3000
    }

    response = requests.post(LLM_API_URL, headers=headers, json=payload, timeout=120)
    response.raise_for_status()
    result = response.json()

    report = result["choices"][0]["message"]["content"]
    returnreport


# ============ 第三步：保存和发送 ============
def save_report(report, output_dir="./reports"):
    """
    把日报保存为Markdown文件
    """
    os.makedirs(output_dir, exist_ok=True)

    date_str = datetime.now().strftime("%Y%m%d")
    filename = f"industry_report_{date_str}.md"
    filepath = os.path.join(output_dir, filename)

    with open(filepath, "w", encoding="utf-8") as f:
        f.write(report)

    print(f"日报已保存到：{filepath}")
    return filepath


def send_email(subject, body):
    """
    通过邮件发送日报（可选功能）
    """
    if not EMAIL_ENABLED:
        print("邮件发送未启用，跳过")
        return

    msg = MIMEMultipart("alternative")
    msg["Subject"] = subject
    msg["From"] = EMAIL_FROM
    msg["To"] = ", ".join(EMAIL_TO)

    # 把Markdown转成简单的HTML（保留基本格式）
    html_body = body.replace("\n", "<br>")
    html_body = f"<div style='font-family: sans-serif; max-width: 800px; margin: 0 auto;'>{html_body}</div>"

    msg.attach(MIMEText(body, "plain", "utf-8"))
    msg.attach(MIMEText(html_body, "html", "utf-8"))

    try:
        with smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT) asserver:
            server.login(EMAIL_FROM, EMAIL_PASSWORD)
            server.sendmail(EMAIL_FROM, EMAIL_TO, msg.as_string())
        print(f"邮件已发送到：{', '.join(EMAIL_TO)}")
    except Exception as e:
        print(f"邮件发送失败：{e}")


# ============ 定时任务 ============
def run_daily_job():
    """
    完整的日报生成流程：爬取 -> 摘要 -> 保存 -> 发送
    """
    print(f"\n{'='*50}")
    print(f"行业日报生成器 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"{'='*50}")

    # 1. 爬取新闻
    print("\n[1/4] 开始爬取新闻...")
    news_list = crawl_all_sources()
    print(f"共抓取到 {len(news_list)} 条新闻")

    if not news_list:
        print("未抓取到任何新闻，请检查网络和新闻源配置")
        return

    # 2. AI生成日报
    print("\n[2/4] 正在调用AI生成日报...")
    report = generate_daily_report(news_list)

    # 3. 保存文件
    print("\n[3/4] 保存日报文件...")
    filepath = save_report(report)

    # 4. 发送邮件
    print("\n[4/4] 发送邮件...")
    date_str = datetime.now().strftime("%m月%d日")
    send_email(f"行业日报 - {date_str}", report)

    print(f"\n{'='*50}")
    print("日报生成完毕")
    print(f"{'='*50}\n")


def schedule_daily_job():
    """
    设置定时任务，每天早上8:30自动执行
    （8:30跑，9:00之前肯定能生成完）
    """
    import schedule

    # 每天早上8:30执行
    schedule.every().day.at("08:30").do(run_daily_job)

    print("定时任务已启动")
    print("每天 08:30 自动生成行业日报")
    print("按 Ctrl+C 停止\n")

    while True:
        schedule.run_pending()
        time.sleep(30)


# ============ 主入口 ============
if __name__ == "__main__":
    import sys

    if len(sys.argv) >1 and sys.argv[1] == "--schedule":
        # 定时模式
        schedule_daily_job()
    else:
        # 单次运行模式（测试用）
        run_daily_job()

==================================================
行业日报生成器 - 2024-12-20 08:30:05
==================================================

[1/4] 开始爬取新闻...
正在爬取：36氪 (https://36kr.com/newsflashes)
  获取到 18 条新闻
正在爬取：物流沙龙 (https://www.logclub.com/news)
  获取到 12 条新闻
正在爬取：亿邦动力 (https://www.ebrun.com/news/)
  获取到 15 条新闻
共抓取到 45 条新闻

[2/4] 正在调用AI生成日报...

[3/4] 保存日报文件...
日报已保存到：./reports/industry_report_20241220.md

[4/4] 发送邮件...
邮件发送未启用，跳过

==================================================
日报生成完毕
==================================================

# 行业日报 - 2024年12月20日

## 今日要点

京东物流宣布将在2025年新增50个智能仓储中心，总投资超过80亿元，进一步扩大其供应链基础设施布局。与此同时，SaaS赛道迎来一笔大额融资，供应链管理平台"链易"完成B轮2亿元融资。政策层面，交通运输部发布了新版《网络货运经营管理暂行办法》，对平台型物流企业的合规要求做了细化。

## 物流与供应链

**京东物流2025年新增50个智能仓储中心**
京东物流披露明年基建计划，80亿投资规模创近三年新高。这释放出一个信号：头部物流企业在降本增效的同时，仍在加码重资产投入，行业集中度会进一步提升。

**菜鸟发布"全球5日达"升级计划**
菜鸟国际快递宣布跨境时效覆盖扩至全球50个核心城市。跨境电商卖家将是最大受益者，物流时效直接影响复购率。

**交通运输部发布网络货运新规**
新规要求平台企业对实际承运人资质做实质性审核，不再是"挂个名就行"。中小网络货运平台合规成本会上升，行业可能迎来一波洗牌。

## SaaS与企业服务

**供应链SaaS"链易"完成B轮2亿融资**
链易主打中小制造企业的采购协同，这轮融资由红杉领投。说明资本对垂直SaaS仍有兴趣，关键看能不能真正解决行业痛点。

**钉钉发布制造业专属解决方案**
钉钉针对制造业场景推出了生产管理、质量追溯等模块。大厂在垂直行业做深，中小SaaS厂商的差异化空间被压缩了。

## 编辑观点

今天最值得关注的信号是物流行业的"两头分化"。一边是京东、菜鸟这样的巨头在基础设施上持续加码，一边是监管对中小平台合规要求收紧。对SaaS创业者来说，机会可能在于：巨头做标准化的重资产，你可以做它们不愿意碰的定制化轻服务，嵌进巨头的生态里活下来。