Spaces:

Tak0000000
/

news-agent-api

Sleeping

App Files Files Community

news-agent-api / app.py

Tak0000000

Upload 2 files

024d5b4 verified 2 months ago

Raw

History Blame Contribute Delete

42.5 kB

	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 裡的 <img>
	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 裡的 <img>
	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 裡的 <img>
	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 (
	"<html><body style='font-family:sans-serif;text-align:center;padding-top:80px'>"
	"<h1>🟢 雲端大腦運行完美，等待手機呼叫！</h1>"
	"<p>Multi-Source AI News Agent v2.0</p>"
	"<p>支援來源：科技新報 \| 癮科技 \| 數位時代 \| Yahoo 新聞 \| HKEPC</p>"
	"</body></html>"
	)


	@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