import os
import json
import random
import re
import time
import xml.etree.ElementTree as ET
from typing import List, Dict, Optional, Tuple
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urlparse
import requests
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from openai import OpenAI
from bs4 import BeautifulSoup
# ─────────────────────────────────────────────────────────────────
# App 初始化
# ─────────────────────────────────────────────────────────────────
app = FastAPI()
# ─────────────────────────────────────────────────────────────────
# 設定
# ─────────────────────────────────────────────────────────────────
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_MODEL = os.getenv("DEEPSEEK_MODEL", "deepseek-chat")
client = OpenAI(api_key=DEEPSEEK_API_KEY, base_url="https://api.deepseek.com/v1")
HEADERS = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/125.0.0.0 Safari/537.36"
),
"Accept-Language": "zh-TW,zh;q=0.9,en;q=0.8",
}
ARTICLES_PER_SOURCE = 10
MAX_FETCH_SECONDS = 12 # 每個來源最多等幾秒
# ─────────────────────────────────────────────────────────────────
# Unsplash 高品質科技/財經圖庫（防呆用）
# ─────────────────────────────────────────────────────────────────
UNSPLASH_POOL = [
# 科技類
"https://images.unsplash.com/photo-1518770660439-4636190af475?w=800&q=80", # 晶片
"https://images.unsplash.com/photo-1504711434969-e33886168f5c?w=800&q=80", # 新聞桌
"https://images.unsplash.com/photo-1488590528505-98d2b5aba04b?w=800&q=80", # 程式碼
"https://images.unsplash.com/photo-1531297484001-80022131f5a1?w=800&q=80", # 筆電
"https://images.unsplash.com/photo-1460925895917-afdab827c52f?w=800&q=80", # 數據分析
"https://images.unsplash.com/photo-1551288049-bebda4e38f71?w=800&q=80", # 儀表板
"https://images.unsplash.com/photo-1550751827-4bd374c3f58b?w=800&q=80", # 資安
"https://images.unsplash.com/photo-1519389950473-47ba0277781c?w=800&q=80", # 辦公室
"https://images.unsplash.com/photo-1451187580459-43490279c0fa?w=800&q=80", # 全球網路
"https://images.unsplash.com/photo-1526374965328-7f61d4dc18c5?w=800&q=80", # 網絡
"https://images.unsplash.com/photo-1558494949-ef010cbdcc31?w=800&q=80", # 伺服器
"https://images.unsplash.com/photo-1560472354-b33ff0c44a43?w=800&q=80", # 數位行銷
"https://images.unsplash.com/photo-1504384308090-c894fdcc538d?w=800&q=80", # 工作空間
"https://images.unsplash.com/photo-1559028012-481c04fa702d?w=800&q=80", # 雲端
"https://images.unsplash.com/photo-1526628953301-3e589a6a8b74?w=800&q=80", # 金融科技
"https://images.unsplash.com/photo-1677442136019-21780ecad995?w=800&q=80", # AI
"https://images.unsplash.com/photo-1511707171634-5f897ff02aa9?w=800&q=80", # 手機
"https://images.unsplash.com/photo-1633356122544-f134324a6cee?w=800&q=80", # 區塊鏈
"https://images.unsplash.com/photo-1581091226825-a39f28c8ff2a?w=800&q=80", # 工程
"https://images.unsplash.com/photo-1485827404703-89b55fcc595e?w=800&q=80", # 機器人
# 財經/商業類
"https://images.unsplash.com/photo-1611974789855-9c2a0a7236a3?w=800&q=80", # 股市
"https://images.unsplash.com/photo-1590283603385-17ffb3a7f3f4?w=800&q=80", # 財經圖表
"https://images.unsplash.com/photo-1553729459-afe8f2e2ed65?w=800&q=80", # 商業大樓
"https://images.unsplash.com/photo-1450101499163-c884c5c54a85?w=800&q=80", # 商業分析
# 旅遊/生活類
"https://images.unsplash.com/photo-1488646953014-85cb44e25828?w=800&q=80", # 旅遊
"https://images.unsplash.com/photo-1507525428034-b723cf961d3e?w=800&q=80", # 海灘
"https://images.unsplash.com/photo-1469854523086-cc02fe5d8800?w=800&q=80", # 旅行
"https://images.unsplash.com/photo-1476514525535-07fb3b4ae5f1?w=800&q=80", # 山景
"https://images.unsplash.com/photo-1444723121867-7a241cacace9?w=800&q=80", # 城市
"https://images.unsplash.com/photo-1533105079780-92b9be482077?w=800&q=80", # 美食
"https://images.unsplash.com/photo-1414235077428-338989a2e8c0?w=800&q=80", # 餐飲
# 材料/工業類
"https://images.unsplash.com/photo-1535813547-99c4740f9fc5?w=800&q=80", # 金屬
"https://images.unsplash.com/photo-1581092160562-40aa08e78837?w=800&q=80", # 工廠
"https://images.unsplash.com/photo-1565122645852-3dd6403dcad2?w=800&q=80", # 材料科學
"https://images.unsplash.com/photo-1504917595217-d4dc5ebe6122?w=800&q=80", # 塑膠/聚合物
# 科學類
"https://images.unsplash.com/photo-1507413245164-6160d8298b31?w=800&q=80", # 科學
"https://images.unsplash.com/photo-1532094349884-543bc11b234d?w=800&q=80", # 實驗室
# 冷知識/趣味類
"https://images.unsplash.com/photo-1533073526757-2c8ca1df9f1c?w=800&q=80", # 好奇
"https://images.unsplash.com/photo-1516410529446-2e27be6d47d6?w=800&q=80", # 知識
]
# ─────────────────────────────────────────────────────────────────
# 新聞來源定義
# ─────────────────────────────────────────────────────────────────
# ── 備用清單（當 news_sources.json 無法讀取時使用）──
FALLBACK_SOURCES: List[Dict] = [
{"name": "科技新報", "type": "rss", "url": "https://technews.tw/feed/"},
{"name": "癮科技", "type": "rss", "url": "https://chinese.engadget.com/rss.xml"},
{"name": "數位時代", "type": "rss", "url": "https://www.bnext.com/rss"},
{"name": "Yahoo 新聞", "type": "rss", "url": "https://hk.news.yahoo.com/rss/"},
{"name": "RTHK 本地新聞", "type": "rss", "url": "https://rthk.hk/rthk/news/rss/c_expressnews_clocal.xml"},
{"name": "RTHK 國際新聞", "type": "rss", "url": "https://rthk.hk/rthk/news/rss/c_expressnews_cinternational.xml"},
{"name": "Gizmodo", "type": "rss", "url": "https://gizmodo.com/rss"},
{"name": "Hackaday", "type": "rss", "url": "https://hackaday.com/feed/"},
]
def load_news_sources() -> List[Dict]:
"""從 news_sources.json 載入新聞來源清單；失敗時回退到備用清單。"""
import os as _os
config_path = _os.path.join(_os.path.dirname(_os.path.abspath(__file__)), "news_sources.json")
try:
with open(config_path, "r", encoding="utf-8") as f:
sources = json.load(f)
if isinstance(sources, list) and len(sources) > 0:
print(f"📋 成功載入 news_sources.json（{len(sources)} 個來源）", flush=True)
return sources
else:
raise ValueError("news_sources.json 格式無效（需為非空陣列）")
except Exception as e:
print(f"⚠️ 無法讀取 news_sources.json：{e}，使用備用清單（{len(FALLBACK_SOURCES)} 個來源）", flush=True)
return FALLBACK_SOURCES
NEWS_SOURCES: List[Dict] = load_news_sources()
# ─────────────────────────────────────────────────────────────────
# 圖片處理輔助函式
# ─────────────────────────────────────────────────────────────────
def extract_image_from_html(html_text: str) -> Optional[str]:
"""從 HTML 內容中提取第一張圖片的網址。"""
if not html_text:
return None
# 方法一：BeautifulSoup 解析
try:
soup = BeautifulSoup(html_text, "html.parser")
img = soup.find("img")
if img:
for attr in ["src", "data-src", "data-original"]:
src = img.get(attr)
if src:
if src.startswith("http"):
return src
if src.startswith("//"):
return "https:" + src
except Exception:
pass
# 方法二：正則表達式兜底
try:
match = re.search(
r'(?:src|data-src|data-original)=["\'](https?://[^"\'>]+?\.(?:jpg|jpeg|png|webp|gif)[^"\'>]*)["\']',
html_text, re.IGNORECASE,
)
if match:
return match.group(1)
except Exception:
pass
return None
# RSS/Atom 常用命名空間
NS = {
"media": "http://search.yahoo.com/mrss/",
"content": "http://purl.org/rss/1.0/modules/content/",
"dc": "http://purl.org/dc/elements/1.1/",
"atom": "http://www.w3.org/2005/Atom",
}
def extract_image_from_rss_item(item: ET.Element) -> Optional[str]:
"""從 RSS/Atom item 元素中提取圖片網址（使用 ElementTree）。"""
# 1) media:content / media:thumbnail
for tag in ["{http://search.yahoo.com/mrss/}content",
"{http://search.yahoo.com/mrss/}thumbnail"]:
media_el = item.find(tag)
if media_el is not None:
url = media_el.get("url", "")
if url.startswith("http"):
return url
# 也可能在所有後代中
for el in item.iter(tag):
url = el.get("url", "")
if url.startswith("http"):
return url
# 2) enclosure (RSS 2.0)
for enc in item.findall("enclosure"):
url = enc.get("url", "")
mime = enc.get("type", "")
if url.startswith("http"):
if "image" in mime or any(
url.lower().endswith(ext) for ext in (".jpg", ".jpeg", ".png", ".webp", ".gif")
):
return url
# 3) content:encoded 裡的 ![]()
content_el = item.find("{http://purl.org/rss/1.0/modules/content/}encoded")
if content_el is not None and content_el.text:
img = extract_image_from_html(content_el.text)
if img:
return img
# 4) description 裡的 ![]()
desc_el = item.find("description")
if desc_el is not None and desc_el.text:
img = extract_image_from_html(desc_el.text)
if img:
return img
# 5) Atom: content 裡的 ![]()
for tag in ["content", "{http://www.w3.org/2005/Atom}content"]:
atom_content = item.find(tag)
if atom_content is not None and atom_content.text:
img = extract_image_from_html(atom_content.text)
if img:
return img
return None
def get_entry_text(item: ET.Element, *tags: str) -> str:
"""從 RSS/Atom item 中取得文字內容，依序嘗試多個 tag（含命名空間變體）。"""
for tag in tags:
# 嘗試純 tag 名稱
el = item.find(tag)
if el is not None and el.text:
return el.text.strip()
# 嘗試帶 Atom 命名空間
el = item.find(f"{{http://www.w3.org/2005/Atom}}{tag}")
if el is not None and el.text:
return el.text.strip()
return ""
def get_fallback_image(title: str = "") -> str:
"""根據新聞關鍵字回傳相關 Unsplash 圖片；無匹配則隨機挑選。"""
kw = title.lower() if title else ""
mapping = [
# 科技
(["ai", "人工智慧", "artificial intelligence", "gpt", "llm", "深度學習", "machine learning"],
"https://images.unsplash.com/photo-1677442136019-21780ecad995?w=800&q=80"),
(["手機", "phone", "iphone", "android", "samsung", "mobile"],
"https://images.unsplash.com/photo-1511707171634-5f897ff02aa9?w=800&q=80"),
(["晶片", "chip", "半導體", "semiconductor", "nvidia", "intel", "amd", "tsmc", "台積電"],
"https://images.unsplash.com/photo-1518770660439-4636190af475?w=800&q=80"),
(["資安", "安全", "security", "cyber", "hack", "漏洞"],
"https://images.unsplash.com/photo-1550751827-4bd374c3f58b?w=800&q=80"),
(["雲端", "cloud", "aws", "azure", "gcp"],
"https://images.unsplash.com/photo-1451187580459-43490279c0fa?w=800&q=80"),
(["數據", "data", "big data", "分析", "analytics"],
"https://images.unsplash.com/photo-1551288049-bebda4e38f71?w=800&q=80"),
(["機器人", "robot", "automation", "自動化"],
"https://images.unsplash.com/photo-1485827404703-89b55fcc595e?w=800&q=80"),
(["電動車", "ev", "tesla", "汽車", "vehicle", "自駕"],
"https://images.unsplash.com/photo-1593941707882-a5bba14938c7?w=800&q=80"),
(["區塊鏈", "blockchain", "crypto", "加密", "bitcoin", "web3"],
"https://images.unsplash.com/photo-1633356122544-f134324a6cee?w=800&q=80"),
# 財經
(["金融", "財經", "finance", "fintech", "bank", "投資", "股票", "股市", "基金", "匯率", "港股", "美股"],
"https://images.unsplash.com/photo-1611974789855-9c2a0a7236a3?w=800&q=80"),
(["經濟", "economy", "貿易", "trade", "gdp", "通膨", "加息", "減息"],
"https://images.unsplash.com/photo-1450101499163-c884c5c54a85?w=800&q=80"),
# 旅遊/生活
(["旅遊", "travel", "旅行", "景點", "酒店", "機票", "美食", "餐廳", "打卡"],
"https://images.unsplash.com/photo-1488646953014-85cb44e25828?w=800&q=80"),
(["深圳", "北上", "大灣區", "灣區", "內地"],
"https://images.unsplash.com/photo-1444723121867-7a241cacace9?w=800&q=80"),
(["生活", "lifestyle", "潮流", "時尚", "購物", "shopping"],
"https://images.unsplash.com/photo-1476514525535-07fb3b4ae5f1?w=800&q=80"),
# 材料/工業
(["金屬", "metal", "鋼鐵", "礦", "原料", "塑膠", "plastic", "材料", "material", " polymer"],
"https://images.unsplash.com/photo-1535813547-99c4740f9fc5?w=800&q=80"),
(["工業", "製造", "工廠", "manufacturing"],
"https://images.unsplash.com/photo-1581092160562-40aa08e78837?w=800&q=80"),
# 科學
(["科學", "science", "研究", "research", "太空", "nasa", "生物", "物理", "化學"],
"https://images.unsplash.com/photo-1507413245164-6160d8298b31?w=800&q=80"),
# 冷知識/趣味
(["冷知識", "趣聞", "奇聞", "歷史", "考古", "謎團", "心理"],
"https://images.unsplash.com/photo-1533073526757-2c8ca1df9f1c?w=800&q=80"),
]
for keywords, url in mapping:
if any(k in kw for k in keywords):
return url
return random.choice(UNSPLASH_POOL)
def ensure_image_url(raw_url: Optional[str], title: str = "") -> str:
"""保證一定回傳有效圖片網址：優先原始網址，否則 Unsplash 補上。"""
if raw_url and raw_url.startswith("http"):
return raw_url
return get_fallback_image(title)
def clean_html(raw_html: str) -> str:
"""將 HTML 轉為純文字，保留基本段落結構。"""
if not raw_html:
return ""
try:
soup = BeautifulSoup(raw_html, "html.parser")
# 移除 script / style
for tag in soup(["script", "style", "nav", "footer", "header", "aside"]):
tag.decompose()
text = soup.get_text(separator="\n")
# 壓縮多餘空白行
text = re.sub(r"\n\s*\n", "\n", text)
text = re.sub(r" +", " ", text)
return text.strip()
except Exception:
# 簡易 regex 去除 HTML tag
text = re.sub(r"<[^>]+>", " ", raw_html)
text = re.sub(r"\s+", " ", text)
return text.strip()
def guess_category(source_name: str) -> str:
"""根據來源名稱推測文章分類標籤（AI 分類失敗時的 fallback）。"""
name = source_name.lower()
if any(k in name for k in ["財經", "aastocks", "鉅亨", "finance", "money", "stock"]):
return "財經"
if any(k in name for k in ["rthk", "國際", "global", "world"]):
return "國際"
if any(k in name for k in ["旅遊", "travel", "utravel", "生活", "ulifestyle", "wave", "wavetv"]):
return "旅遊"
if any(k in name for k in ["01", "大灣區", "北上", "深圳"]):
return "旅遊"
if any(k in name for k in ["科學", "science", "tomorrowsci", "地球圖輯", "冷知識"]):
return "科學"
if any(k in name for k in ["plastic", "metal", "azom", "材料", "工業"]):
return "材料工業"
if any(k in name for k in ["trend", "熱搜", "google trend"]):
return "熱門趨勢"
if any(k in name for k in ["gizmodo", "hackaday", "科技新報", "癮科技", "數位時代", "inside", "qbit", "hkepc"]):
return "科技"
return "科技" # 預設
# ─────────────────────────────────────────────────────────────────
# 新聞來源擷取函式
# ─────────────────────────────────────────────────────────────────
def fetch_rss_articles(source: Dict) -> List[Dict]:
"""透過 RSS feed 抓取最新文章（最多 ARTICLES_PER_SOURCE 篇）。
支援 RSS 2.0 與 Atom 格式，使用內建 xml.etree.ElementTree 解析。"""
name = source["name"]
url = source["url"]
articles = []
print(f"📡 [{name}] 開始抓取 RSS: {url}", flush=True)
try:
resp = requests.get(url, headers=HEADERS, timeout=MAX_FETCH_SECONDS)
resp.raise_for_status()
resp.encoding = resp.apparent_encoding or "utf-8"
xml_text = resp.text
except Exception as e:
print(f"⚠️ [{name}] RSS 連線失敗: {e}", flush=True)
return articles
# 嘗試解析 XML
try:
root = ET.fromstring(xml_text)
except ET.ParseError as e:
print(f"⚠️ [{name}] XML 解析失敗: {e}", flush=True)
return articles
# 判斷 RSS 2.0 或 Atom 格式
items = []
# RSS 2.0: channel > item
rss_items = root.findall(".//item")
if rss_items:
items = rss_items
else:
# Atom: feed > entry
atom_items = root.findall(".//{http://www.w3.org/2005/Atom}entry")
if not atom_items:
atom_items = root.findall(".//entry")
items = atom_items
if not items:
print(f"⚠️ [{name}] 找不到任何新聞條目", flush=True)
return articles
items = items[:ARTICLES_PER_SOURCE]
print(f"📰 [{name}] 取得 {len(items)} 篇文章", flush=True)
for item in items:
# 標題
title = (
get_entry_text(item, "title") or
get_entry_text(item, "{http://www.w3.org/2005/Atom}title")
)
if not title:
continue
# 內文：優先 content:encoded，其次 description / summary / content
content_text = get_entry_text(
item,
"{http://purl.org/rss/1.0/modules/content/}encoded",
"description",
"summary",
"content",
)
# 清理 HTML 標籤
clean_content = clean_html(content_text)
# 限制內文長度（避免 token 爆炸）
if len(clean_content) > 600:
clean_content = clean_content[:600] + "..."
# 圖片提取
image_url = extract_image_from_rss_item(item)
image_url = ensure_image_url(image_url, title)
articles.append({
"title": title,
"content": clean_content or title,
"image_url": image_url,
"source": name,
})
return articles
def fetch_html_articles(source: Dict) -> List[Dict]:
"""從一般 HTML 網頁抓取新聞列表（例如 HKEPC）。"""
name = source["name"]
url = source["url"]
articles = []
print(f"🌐 [{name}] 開始爬取 HTML: {url}", flush=True)
try:
resp = requests.get(url, headers=HEADERS, timeout=MAX_FETCH_SECONDS)
resp.raise_for_status()
resp.encoding = resp.apparent_encoding or "utf-8"
soup = BeautifulSoup(resp.text, "html.parser")
except Exception as e:
print(f"⚠️ [{name}] HTML 連線失敗: {e}", flush=True)
return articles
# ── HKEPC 專用選擇器 ──
# 嘗試多種可能的文章區塊選擇器（依優先序）
candidates = []
# 常見新聞列表選擇器
selectors = [
"a[href*='view']", # HKEPC 新聞連結格式
".news-item a",
".article-list a",
".post-title a",
".entry-title a",
"article a",
".content a[href*='news']",
".main-content a",
"h2 a", "h3 a", "h4 a", # 標題連結
]
seen_urls = set()
for selector in selectors:
try:
for tag in soup.select(selector):
href = tag.get("href", "")
text = tag.get_text(strip=True)
if not href or not text or len(text) < 6:
continue
# 過濾掉非新聞連結
if any(skip in href.lower() for skip in ["login", "register", "ad.", "banner", "javascript:"]):
continue
if href in seen_urls:
continue
seen_urls.add(href)
candidates.append({"title": text, "link": href})
except Exception:
continue
if len(candidates) >= ARTICLES_PER_SOURCE:
break
print(f"🔗 [{name}] 找到 {len(candidates)} 個候選連結", flush=True)
# 擷取前 N 篇
for item in candidates[:ARTICLES_PER_SOURCE]:
title = item["title"]
link = item["link"]
# 補全相對路徑
if link.startswith("/"):
link = "https://www.hkepc.com" + link
elif not link.startswith("http"):
link = "https://www.hkepc.com/" + link
content_text = title # 預設用標題當內容
image_url = None
# 嘗試進入文章內頁抓取更多內容
try:
art_resp = requests.get(link, headers=HEADERS, timeout=6)
art_resp.raise_for_status()
art_resp.encoding = art_resp.apparent_encoding or "utf-8"
art_soup = BeautifulSoup(art_resp.text, "html.parser")
# 內文
for art_sel in ["article", ".article-content", ".content", ".post-content", ".entry-content", "main", ".news-content"]:
body = art_soup.select_one(art_sel)
if body:
content_text = clean_html(str(body))
break
if len(content_text) > 600:
content_text = content_text[:600] + "..."
# 圖片
page_html = str(art_soup)
image_url = extract_image_from_html(page_html)
except Exception as e:
print(f" ⚠️ 無法抓取內頁 {link[:60]}: {e}", flush=True)
image_url = ensure_image_url(image_url, title)
articles.append({
"title": title,
"content": content_text or title,
"image_url": image_url,
"source": name,
})
print(f"📰 [{name}] 擷取 {len(articles)} 篇文章", flush=True)
return articles
def fetch_articles_from_source(source: Dict) -> List[Dict]:
"""統一入口：根據來源類型呼叫對應的擷取函式；失敗時自動切換 HTML 備用爬蟲。"""
articles = []
try:
if source["type"] == "rss":
articles = fetch_rss_articles(source)
elif source["type"] == "html":
articles = fetch_html_articles(source)
else:
print(f"⚠️ 未知來源類型: {source['type']}", flush=True)
except Exception as e:
print(f"❌ [{source['name']}] 擷取異常: {e}", flush=True)
# ── 若正常抓取失敗（0 篇），自動啟動 HTML 備用爬蟲 ──
if not articles:
print(f"🔄 [{source['name']}] 正常抓取取得 0 篇，觸發 HTML 備用爬蟲...", flush=True)
articles = fetch_html_fallback(source)
return articles
# ─────────────────────────────────────────────────────────────────
# DeepSeek AI 摘要處理
# ─────────────────────────────────────────────────────────────────
SYSTEM_PROMPT = """你是一個專業的繁體中文新聞編輯秘書。你將收到一批新聞文章（JSON 陣列），
每篇文章包含 title、content、image_url、source 四個欄位。
請為每一篇文章完成以下任務：
1. **語言判斷**：先判斷原文 title 和 content 的主要語言。
- 若為**中文**：直接產出繁體中文的 title 和 summary（單語模式，摘要 80～200 字）。
- 若為**英文或其他外語**：自動切換為「中英對照模式」（見第 2 點）。
2. **中英對照模式**（僅適用於外語原文）：
title 和 summary 必須用換行符號 `\n` 分隔上下兩半段：
- **上半段**：流暢、道地的繁體中文翻譯（使用香港/台灣常用表達方式），摘要 80～200 字。
- **下半段**：對應的英文原文（title 對照英文原標題；summary 對照由 AI 生成的流暢英文摘要，40～70 個英文字）。
對照模式輸出範例：
{
"title": "記憶體晶片短缺無解\nMemory Chip Shortage Remains Unresolved",
"summary": "由於 AI 高價訂單增加，消費性電子業將面臨嚴重的晶片缺貨衝擊。\nDue to the surge in high-priced AI orders, the consumer electronics industry will face a severe chip shortage impact.",
"category": "科技",
"image_url": "https://...",
"source": "..."
}
3. **分類標籤**：根據文章內容，從以下 9 個選項中挑選最貼近的一個 category：
科技 | 財經 | 國際 | 旅遊 | 生活 | 科學 | 材料工業 | 冷知識 | 熱門趨勢
4. 保留原本的 image_url（不要更改）。
5. 保留原本的 source（不要更改）。
6. 保留原本的 title（中文模式可微調標點符號；對照模式上半段為翻譯、下半段保留英文原標題）。
你必須**嚴格回傳一個 JSON 陣列**，格式如下：
[
{
"title": "繁體中文標題（外語時：中文\nEnglish）",
"summary": "80～200字繁體中文摘要（外語時：中文摘要\nEnglish summary）",
"category": "科技",
"image_url": "https://...",
"source": "來源名稱"
},
...
]
注意事項：
- 中文模式摘要必須在 80～200 個中文字之間，不可過短，也不可超過 200 字
- 對照模式上半段繁體中文摘要同樣 80～200 字，下半段英文摘要 40～70 個英文字
- category 必須是列出的 9 個選項之一，不可自行發明
- 所有繁體中文必須使用 zh-TW 習慣用語
- 不可遺漏任何一篇文章
- 不可新增不存在的新聞
- 回傳的 JSON 陣列長度必須等於輸入的文章數量"""
# ── HTML 備用爬蟲專用提示詞 ──
HTML_FALLBACK_SYSTEM_PROMPT = """你是一個精準的網頁新聞提煉專家。我將提供一段從網站首頁擷取下來的 HTML 純文字內容（已移除 script/style 標籤）。
這段文字混雜了選單、廣告、頁尾與真正的新聞條目。
你的任務是：
1. 從這段雜亂的文字中，精確找出**最新的 2 篇重要新聞**
2. 自行判斷原文語言，遵循中英雙語對照規則（與主新聞秘書相同）：
- 外語原文 → 上半段繁體中文 / 下半段英文對照
- 中文原文 → 僅繁體中文
3. 繁體中文摘要 80～200 字，英文摘要 40～70 字
4. 為每篇新聞指定一個 category（從：科技 | 財經 | 國際 | 旅遊 | 生活 | 科學 | 材料工業 | 冷知識 | 熱門趨勢 中挑選）
5. 圖片欄位（image_url）若無法從文本中取得，請留空字串 ""
你必須**嚴格回傳一個 JSON 物件**，內含 "articles" 陣列：
{
"articles": [
{
"title": "...",
"summary": "...",
"category": "...",
"image_url": "",
"source": "提供的網站名稱"
},
...
]
}
注意：只回傳真正的新聞內容，忽略導覽選單、頁尾連結、廣告、社交媒體按鈕等雜訊。"""
def fetch_html_fallback(source: Dict) -> List[Dict]:
"""
HTML 備用爬蟲：
當 RSS/正常爬蟲失敗時，抓取網站首頁並請 DeepSeek 直接從雜亂文本中提取新聞。
回傳的文章已具備最終格式（含 title, summary, category, image_url, source）。
"""
name = source["name"]
url = source["url"]
# 從 URL 提取首頁網址
try:
parsed = urlparse(url)
base_url = f"{parsed.scheme}://{parsed.netloc}"
except Exception:
base_url = url
print(f"🔄 [{name}] RSS 失敗，啟動 HTML 備用爬蟲 → {base_url}", flush=True)
try:
resp = requests.get(base_url, headers=HEADERS, timeout=MAX_FETCH_SECONDS)
resp.raise_for_status()
resp.encoding = resp.apparent_encoding or "utf-8"
except Exception as e:
print(f" ❌ [{name}] HTML 備用連線失敗: {e}", flush=True)
return []
# ── 提取 body 純文字，限制 10000 字元 ──
try:
soup = BeautifulSoup(resp.text, "html.parser")
# 移除 script / style / nav / footer
for tag in soup(["script", "style", "nav", "footer", "header", "aside"]):
tag.decompose()
if soup.body:
text = soup.body.get_text(separator="\n", strip=True)
else:
text = soup.get_text(separator="\n", strip=True)
# 壓縮空白
text = re.sub(r"\n\s*\n", "\n", text)
text = re.sub(r" +", " ", text)
text = text[:10000]
print(f" 📄 [{name}] 提取純文字 {len(text)} 字元，餵給 DeepSeek...", flush=True)
except Exception as e:
print(f" ❌ [{name}] HTML 文本提取失敗: {e}", flush=True)
return []
if len(text) < 100:
print(f" ⚠️ [{name}] HTML 文本過短，跳過", flush=True)
return []
# ── 餵給 DeepSeek 提取新聞 ──
try:
response = client.chat.completions.create(
model=DEEPSEEK_MODEL,
messages=[
{"role": "system", "content": HTML_FALLBACK_SYSTEM_PROMPT},
{"role": "user", "content": f"網站名稱：{name}\n網址：{base_url}\n\nHTML 文本內容：\n{text}"},
],
response_format={"type": "json_object"},
temperature=0.3,
max_tokens=4096,
)
raw = response.choices[0].message.content
result = json.loads(raw)
# 提取 articles 陣列
articles = []
if isinstance(result, dict):
for key in ["articles", "news", "items", "results", "data"]:
if key in result and isinstance(result[key], list):
articles = result[key]
break
# 若只有單個 key 是 list
if not articles:
for val in result.values():
if isinstance(val, list):
articles = val
break
elif isinstance(result, list):
articles = result
# ── 補完欄位 ──
completed = []
for item in articles[:5]: # 最多取 5 篇
if not isinstance(item, dict):
continue
completed.append({
"title": str(item.get("title", "未知標題")).strip(),
"summary": str(item.get("summary", item.get("content", "暫無摘要"))).strip(),
"category": str(item.get("category", guess_category(name))).strip(),
"image_url": item.get("image_url") or get_fallback_image(str(item.get("title", ""))),
"source": name,
"_fallback": True, # 標記為備用爬蟲產出（已含摘要，不需再次 summarize）
})
if completed:
print(f" ✅ [{name}] HTML 備用爬蟲成功提取 {len(completed)} 篇新聞", flush=True)
else:
print(f" ⚠️ [{name}] HTML 備用爬蟲未找到新聞", flush=True)
return completed
except json.JSONDecodeError as e:
print(f" ❌ [{name}] DeepSeek JSON 解析失敗: {e}", flush=True)
except Exception as e:
print(f" ❌ [{name}] HTML 備用爬蟲 DeepSeek 呼叫失敗: {e}", flush=True)
return []
def summarize_by_deepseek(articles: List[Dict]) -> List[Dict]:
"""將文章餵給 DeepSeek 進行摘要處理，回傳結構化 JSON 陣列。"""
if not articles:
return []
# 只傳必要欄位給 AI，節省 token
input_data = [
{
"title": a["title"],
"content": a.get("content", a["title"]),
"image_url": a["image_url"],
"source": a["source"],
}
for a in articles
]
payload = json.dumps(input_data, ensure_ascii=False)
print(f"🤖 發送 {len(articles)} 篇文章給 DeepSeek 摘要...", flush=True)
try:
response = client.chat.completions.create(
model=DEEPSEEK_MODEL,
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": payload},
],
response_format={"type": "json_object"},
temperature=0.3,
max_tokens=16384,
)
raw = response.choices[0].message.content
print(f"✅ DeepSeek 回傳成功 ({len(raw)} 字元)", flush=True)
# 嘗試解析 JSON
result = json.loads(raw)
# 如果 AI 回傳的是 { "articles": [...] } 之類的包裝物件，自動拆開
if isinstance(result, dict):
for key in ["articles", "news", "items", "results", "data"]:
if key in result and isinstance(result[key], list):
result = result[key]
break
# 如果還是 dict 且只有一個 key 是 list
if isinstance(result, dict):
for val in result.values():
if isinstance(val, list) and len(val) == len(articles):
result = val
break
if not isinstance(result, list):
print(f"⚠️ AI 回傳格式非陣列（型態: {type(result).__name__}），嘗試修復", flush=True)
# 最後手段：逐項重建
result = []
# ── 欄位補完機制 ──
# 合法的分類標籤
valid_categories = {"科技", "財經", "國際", "旅遊", "生活", "科學", "材料工業", "冷知識", "熱門趨勢"}
completed = []
for i, item in enumerate(result[:len(articles)]):
if not isinstance(item, dict):
continue
original = articles[i] if i < len(articles) else {}
# 分類智慧猜測：從來源名稱推斷
cat = str(item.get("category", "")).strip()
if cat not in valid_categories:
src = str(item.get("source", original.get("source", "")))
cat = guess_category(src)
completed.append({
"title": str(item.get("title", original.get("title", "未知標題"))).strip(),
"summary": str(item.get("summary", item.get("content", original.get("content", "暫無摘要")))).strip(),
"category": cat,
"image_url": item.get("image_url") or original.get("image_url", get_fallback_image("")),
"source": str(item.get("source", original.get("source", "未知來源"))).strip(),
})
# 如果 AI 漏掉了一些文章，用原始資料補齊
if len(completed) < len(articles):
print(f"⚠️ AI 只回傳 {len(completed)}/{len(articles)} 篇，補齊中...", flush=True)
for i in range(len(completed), len(articles)):
a = articles[i]
completed.append({
"title": a["title"],
"summary": clean_html(a.get("content", a["title"]))[:200],
"category": guess_category(a.get("source", "")),
"image_url": a["image_url"],
"source": a["source"],
})
print(f"📋 最終輸出 {len(completed)} 篇新聞", flush=True)
return completed
except json.JSONDecodeError as e:
print(f"❌ DeepSeek 回傳的 JSON 無法解析: {e}", flush=True)
try:
print(f" Raw (前 500 字): {raw[:500]}", flush=True)
except NameError:
print(" Raw: (無法取得原始回應)", flush=True)
except Exception as e:
print(f"❌ DeepSeek 呼叫失敗: {e}", flush=True)
# ── Fallback：直接回傳原始文章（不做 AI 摘要）──
print("🔄 降級：使用原始文章資料回傳", flush=True)
fallback = []
for a in articles:
fallback.append({
"title": a["title"],
"summary": clean_html(a.get("content", a["title"]))[:200],
"category": guess_category(a.get("source", "")),
"image_url": a["image_url"],
"source": a["source"],
})
return fallback
# ─────────────────────────────────────────────────────────────────
# API Endpoints
# ─────────────────────────────────────────────────────────────────
@app.get("/", response_class=HTMLResponse)
@app.head("/")
def home():
"""Hugging Face 健康檢查頁面"""
return (
""
"🟢 雲端大腦運行完美，等待手機呼叫！
"
"Multi-Source AI News Agent v2.0
"
"支援來源：科技新報 | 癮科技 | 數位時代 | Yahoo 新聞 | HKEPC
"
""
)
@app.get("/get_news")
def get_news():
"""
主新聞端點：
1. 從 news_sources.json 載入的所有來源平行抓取最新新聞
2. RSS/HTML 失敗時自動切換 HTML 備用爬蟲（DeepSeek 直接提取）
3. 確保每篇都有圖片
4. 一般文章餵給 DeepSeek 做繁體中文摘要（備用文章已含摘要則略過）
5. 回傳統一 JSON 陣列
"""
start_time = time.time()
all_articles: List[Dict] = []
# ── 步驟 1：平行抓取所有來源 ──
print(f"\n{'='*60}", flush=True)
print(f"🚀 開始抓取 {len(NEWS_SOURCES)} 個新聞來源...", flush=True)
with ThreadPoolExecutor(max_workers=min(len(NEWS_SOURCES), 5)) as executor:
future_map = {
executor.submit(fetch_articles_from_source, src): src["name"]
for src in NEWS_SOURCES
}
for future in as_completed(future_map):
name = future_map[future]
try:
articles = future.result()
all_articles.extend(articles)
print(f"✅ [{name}] 完成，取得 {len(articles)} 篇", flush=True)
except Exception as e:
print(f"❌ [{name}] 執行失敗: {e}", flush=True)
fetch_time = time.time() - start_time
print(f"📦 總共收集 {len(all_articles)} 篇文章（耗時 {fetch_time:.1f}s）", flush=True)
if not all_articles:
return {
"error": "所有新聞來源暫時無法取得，請稍後再試",
"articles": [],
}
# ── 步驟 2：分離「備用爬蟲文章」（已含 DeepSeek 摘要）與「一般文章」──
regular_articles: List[Dict] = []
fallback_articles: List[Dict] = []
for a in all_articles:
if a.pop("_fallback", False):
fallback_articles.append(a)
else:
regular_articles.append(a)
fb_count = len(fallback_articles)
reg_count = len(regular_articles)
print(f"📊 一般文章 {reg_count} 篇 + 備用爬蟲文章 {fb_count} 篇（已摘要，略過處理）", flush=True)
# ── 步驟 3：一般文章餵給 DeepSeek 摘要 ──
BATCH_SIZE = 30
final_articles: List[Dict] = list(fallback_articles) # 備用文章直接加入
if regular_articles:
if len(regular_articles) <= BATCH_SIZE:
final_articles.extend(summarize_by_deepseek(regular_articles))
else:
for i in range(0, len(regular_articles), BATCH_SIZE):
batch = regular_articles[i:i + BATCH_SIZE]
print(f"📦 處理批次 {i // BATCH_SIZE + 1}/{(len(regular_articles) + BATCH_SIZE - 1) // BATCH_SIZE} ({len(batch)} 篇)...", flush=True)
final_articles.extend(summarize_by_deepseek(batch))
total_time = time.time() - start_time
print(f"🏁 全部完成：{len(final_articles)} 篇新聞（一般 {reg_count} + 備用 {fb_count}）（總耗時 {total_time:.1f}s）", flush=True)
print(f"{'='*60}\n", flush=True)
return final_articles