"""
Phase 1: 并发抓取所有军事新闻源的头条列表
Phase 2: 并发获取文章详情摘要
| 运行方式: python3 geopolitics-daily-scan.py
| 输出: stdout（cron 自动投递到微信）
依赖: 标准库 only (urllib, re, json, concurrent.futures, ssl)
"""
import urllib.request
import urllib.error
import json
import re
import ssl
from datetime import datetime, timezone, timedelta
from concurrent.futures import ThreadPoolExecutor, as_completed
import sys
DATE_OVERRIDE = sys.argv[1] if len(sys.argv) > 1 else None
TODAY = DATE_OVERRIDE or datetime.now(timezone(timedelta(hours=8))).strftime('%Y-%m-%d')
BJ_TZ = timezone(timedelta(hours=8))
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
def fetch_url(url, timeout=20, headers=None):
default_headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/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',
}
if headers:
default_headers.update(headers)
req = urllib.request.Request(url, headers=default_headers)
try:
with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
data = resp.read()
charset = resp.headers.get_content_charset() or 'utf-8'
return data.decode(charset, errors='replace')
except Exception:
return None
# ====== 各源采集函数 ======
def fetch_huanqiu():
"""环球网军事 - textarea解析（最可靠）"""
url = "https://mil.huanqiu.com/"
content = fetch_url(url)
if not content:
return []
ids = re.findall(r'<textarea class="item-aid">([^<]+)</textarea>', content)
titles = re.findall(r'<textarea class="item-title">([^<]+)</textarea>', content)
hosts = re.findall(r'<textarea class="item-cnf-host">([^<]+)</textarea>', content)
results = []
for i in range(len(ids)):
aid = ids[i].strip()
title = titles[i].strip() if i < len(titles) else ''
host = hosts[i].strip() if i < len(hosts) else ''
if not title:
continue
article_url = f"https://{host}/article/{aid}" if host and aid else ''
results.append({'source': '环球网军事', 'title': title, 'url': article_url, 'summary': ''})
return results
def fetch_guancha():
"""观察者网军事 - HTML解析"""
url = "https://www.guancha.cn/military-affairs"
content = fetch_url(url)
if not content:
return []
results = []
links = re.findall(r'href="(/military-affairs/\d{4}_\d{2}_\d{2}_[^"]*\.shtml)"', content)
seen = set()
for link in links:
full_url = f"https://www.guancha.cn{link}"
if full_url in seen:
continue
seen.add(full_url)
title_match = re.search(rf'href="{re.escape(link)}"[^>]*>([^<]+)</a>', content)
title = title_match.group(1).strip() if title_match else ''
title = re.sub(r'<[^>]+>', '', title).strip()
if not title or len(title) < 6 or title.isdigit():
continue
results.append({'source': '观察者网军事', 'title': title, 'url': full_url, 'summary': ''})
return results
def fetch_ifeng():
"""凤凰网军事（不稳定，可能返回空）"""
urls = ["https://news.ifeng.com/mil/", "https://mil.ifeng.com/"]
for url in urls:
content = fetch_url(url)
if content and len(content) > 500:
results = []
links = re.findall(
r'<a[^>]*href="(https?://[^"]*ifeng[^"]*/(?:c|article)/[^"]+)"[^>]*>([^<]+)</a>',
content,
)
for href, title in links:
title = re.sub(r'<[^>]+>', '', title).strip()
if title and len(title) > 5:
results.append({'source': '凤凰网军事', 'title': title, 'url': href, 'summary': ''})
if results:
return results
return []
def fetch_sina():
"""新浪军事 - 备选Feed API"""
urls = [
"https://feed.mix.sina.com.cn/api/roll/get?pageid=153&lid=2514&num=20",
"https://interface.sina.cn/pc_api/public_news_data.d.json?cre=tianyi&mod=pcmil&cids=57922&length=15",
]
for url in urls:
content = fetch_url(url)
if not content:
continue
try:
data = json.loads(content)
items = data.get('result', {}).get('data', []) or data.get('data', [])
if not items or items is None:
continue
results = []
for item in items:
title = item.get('title', '').strip()
link = item.get('url', '').strip()
if not title:
continue
if link and not link.startswith('http'):
link = f'https:{link}' if link.startswith('//') else link
results.append({'source': '新浪军事', 'title': title, 'url': link, 'summary': ''})
if results:
return results
except (json.JSONDecodeError, KeyError):
continue
return []
# ====== 文章详情获取 ======
def fetch_summary(url):
content = fetch_url(url, timeout=15)
if not content:
return ''
for pattern in [
r'<meta[^>]*property="og:description"[^>]*content="([^"]+)"',
r'<meta[^>]*name="description"[^>]*content="([^"]+)"',
r'<meta[^>]*property="description"[^>]*content="([^"]+)"',
r'<p[^>]*class="[^"]*summary[^"]*"[^>]*>([^<]+)</p>',
r'<p[^>]*>([^<]{40,})</p>',
]:
m = re.search(pattern, content, re.I)
if m:
return m.group(1).strip().replace(' ', ' ').replace('&', '&')[:200]
return ''
# ====== 分类规则 ======
CLASSIFICATION_RULES = [
('中美关系 / 台海 / 南海',
['中美', '拜登', '台海', '台湾', '南海', '美日', '美菲', '印太', '对华', '美台', '统一', '台独', '两岸',
'中国海军', '越南', '访问']),
('中东局势',
['中东', '伊朗', '以色列', '加沙', '哈马斯', '真主党', '巴勒斯坦', '胡塞', '也门', '叙利亚', '黎巴嫩',
'红海', '波斯湾', '内塔尼亚胡', '斡旋', '卡塔尔', '阿联酋', '霍尔木兹', '布什尔']),
('俄乌冲突',
['俄乌', '俄罗斯', '乌克兰', '普京', '泽连斯基', '基辅', '莫斯科', '顿巴斯', '俄军', '乌军',
'特别军事行动', '远程打击', '停火', '俄黑海']),
('欧洲 / 北约',
['欧洲', '北约', '欧盟', '法国', '德国', '英国', '马克龙', '朔尔茨', '欧洲防务', '北约峰会',
'波兰', '波罗的海', '斯洛伐克', '瑞典', '萨博', '鹰狮']),
('亚太 / 朝鲜半岛',
['朝鲜', '韩国', '日本', '岸田', '金正恩', '朝韩', '半岛', '美日韩', '澳大利亚', '印度',
'东盟', '菲律宾', '导弹试射', '朝核', '文身']),
('核武器 / 军控',
['核武', '核弹', '核威慑', '军控', '核裁军', '核试验', '洲际导弹', '高超音速', '导弹防御',
'战略武器', '核潜艇', '潜射']),
]
def classify(title, summary):
text = (title + ' ' + summary).lower()
scores = [(sum(1 for kw in kws if kw.lower() in text), cat) for cat, kws in CLASSIFICATION_RULES]
scores = [(s, c) for s, c in scores if s > 0]
if scores:
scores.sort(reverse=True)
return scores[0][1]
return '其他国际'
# ====== 主流程 ======
def main():
print("=" * 60)
print(f"国际地缘政治日报 · 采集 {TODAY}")
print("=" * 60)
# Phase 1: 并发抓取
sources = {
'环球网军事': fetch_huanqiu,
'观察者网军事': fetch_guancha,
'凤凰网军事': fetch_ifeng,
'新浪军事': fetch_sina,
}
all_articles = []
with ThreadPoolExecutor(max_workers=4) as ex:
futures = {ex.submit(fn): name for name, fn in sources.items()}
for f in as_completed(futures):
name = futures[f]
try:
arts = f.result()
all_articles.extend(arts)
print(f" ✅ {name}: {len(arts)}条")
except Exception as e:
print(f" ❌ {name}: {e}")
# 去重
seen = set()
unique = []
for a in all_articles:
key = re.sub(r'[^\u4e00-\u9fff\w]', '', a['title'][:30])
if key and key not in seen:
seen.add(key)
unique.append(a)
total_raw = len(all_articles)
print(f"\n📊 原始: {total_raw}条 → 去重后: {len(unique)}条")
# 过滤旧文章（新浪Feed API可能返回多年前的旧新闻）
filtered = []
old_count = 0
for a in unique:
if any(f"/{y}" in a['url'] for y in ["2022", "2023", "2024"]):
old_count += 1
else:
filtered.append(a)
print(f"🗑️ 过滤掉 {old_count} 条旧文章, 剩余 {len(filtered)} 条")
if not filtered:
print("❌ 无可用新闻")
return
# Phase 2: 获取摘要
print(f"\n📝 获取 {min(len(filtered), 40)} 篇文章摘要...")
with ThreadPoolExecutor(max_workers=8) as ex:
futs = {ex.submit(fetch_summary, a['url']): i for i, a in enumerate(filtered[:40])}
for f in as_completed(futs):
i = futs[f]
try:
filtered[i]['summary'] = f.result() or ''
except Exception:
pass
# 分类
classified = {}
for a in filtered:
cat = classify(a['title'], a.get('summary', ''))
classified.setdefault(cat, []).append(a)
# 生成日报（输出到 stdout，cron 自动投递到微信）
category_order = [
'中美关系 / 台海 / 南海',
'中东局势',
'俄乌冲突',
'欧洲 / 北约',
'亚太 / 朝鲜半岛',
'核武器 / 军控',
'其他国际',
]
lines = [f"# 国际地缘政治日报 · {TODAY}\n"]
total_news = 0
for cat in category_order:
items = classified.get(cat, [])
if not items:
continue
lines.append(f"## {cat}\n")
for a in items:
s = a.get('summary', '')
entry = f"**{cat}** | [{a['title']}]({a['url']})"
if s:
entry += f" — {s}"
lines.append(entry)
total_news += 1
lines.append('')
now_str = datetime.now(BJ_TZ).strftime('%Y-%m-%d %H:%M')
lines.append(
f"---\n*采集时间: {now_str} | 数据来源: 环球网军事、观察者网军事、凤凰网军事、新浪军事*\n"
)
# ✅ 直接输出到 stdout（不写本地 .md 文件）
report = '\n'.join(lines)
print("\n" + "=" * 60)
print("📋 国际地缘政治日报（推送版）")
print("=" * 60)
print(report)
print(f"\n📊 共 {len(filtered)} 条新闻, 分布:")
for cat in category_order:
items = classified.get(cat, [])
if items:
print(f" - {cat}: {len(items)}条")
if __name__ == '__main__':
main()