commit 685231fb562993cee1d7c3ae2807162eef5a7e0f Author: lhr Date: Sat Apr 11 22:02:38 2026 +0800 first commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bf37e1d --- /dev/null +++ b/.gitignore @@ -0,0 +1,20 @@ +# Python +__pycache__/ +*.py[cod] +*.pyc +.venv/ + +# 敏感配置 +config.py + +# 状态文件（包含邮件配置等敏感信息） +.ping_state/ +*.json + +# IDE +.vscode/ +.idea/ + +# 系统 +.DS_Store +*.log diff --git a/README.md b/README.md new file mode 100644 index 0000000..821c3b8 --- /dev/null +++ b/README.md @@ -0,0 +1,86 @@ +# 太阳公公监控 + +监控局域网设备上线，配合 AI 生成个性化早安邮件。 + +## 功能 + +- **Ping 监控**：监控指定局域网 IP 是否在线 +- **首次上线通知**：目标设备首次上线时，发送"太阳公公出来了"邮件，包含具体上线时间 +- **未到提醒**：每天 8:15 前目标仍未上线，发送"太阳公公还在睡觉"邮件 +- **AI 早安文案**：三段式早安邮件 + - 第一段：太阳公公状态（AI 单独生成） + - 第二段：今日天气（AI 生成，自然衔接） + - 第三段：今日热搜（AI 找关联，像说相声一样串联成自然段落） +- **每日重置**：状态文件每日自动重置，避免重复发送 + +## 运行时间 + +每天凌晨 2:00 至晚上 22:00 自动监控 + +## 文件结构 + +``` +TaiYangGongGong/ +├── ping_monitor.py # 主脚本（常驻运行） +├── test_ping.py # 测试脚本（单次测试） +├── config.py # 敏感配置（API密钥、邮件账号等）- 不上传 +└── .ping_state/ # 状态文件目录 - 不上传 + ├── state.json + ├── weather_cache.json + └── news_cache.json +``` + +## 首次使用 + +1. 复制配置模板并填入你的信息： + +```bash +cp config.py config.py.bak +# 然后编辑 config.py，填入你的 API 密钥和邮件配置 +``` + +2. 配置 `config.py`： + +```python +# DeepSeek API +DEEPSEEK_API_KEY = "your_api_key" + +# 邮件配置 +SMTP_SERVER = "smtp.qq.com" +SMTP_PORT = 587 +SMTP_USER = "your_email@qq.com" +SMTP_PASSWORD = "your授权码" +FROM_NAME = "月亮姐姐" +TO_EMAILS = ["target_email@example.com"] +``` + +3. 设置系统时区（邮件时间戳用系统时间）： + +```bash +sudo timedatectl set-timezone Asia/Shanghai +sudo ntpdate ntp.aliyun.com # 同步网络时间 +``` + +4. 安装依赖并运行： + +```bash +pip install requests +python test_ping.py # 测试 +python ping_monitor.py # 正式运行 +``` + +## 部署 + +```bash +# 添加到 crontab，每天凌晨2点自动启动 +crontab -e + +# 添加： +0 2 * * * cd /home/lhr/storage/TaiYangGongGong && python ping_monitor.py >> ping.log 2>&1 & +``` + +## 依赖 + +```bash +pip install requests +``` diff --git a/ping_monitor.py b/ping_monitor.py new file mode 100644 index 0000000..1dbcc71 --- /dev/null +++ b/ping_monitor.py @@ -0,0 +1,561 @@ +#!/usr/bin/env python3 +""" +LAN IP 在线监控脚本 +- 运行时间: 每天凌晨2点到晚上10点 +- 监控多个固定局域网IP +- 首次上线发邮件通知 +- 早上8:15前未上线则发送"未到"通知 +""" + +import os +import time +import datetime +import subprocess +import smtplib +import json +import re +import urllib.parse +import urllib.request +from email.mime.text import MIMEText +from email.mime.multipart import MIMEMultipart +from email.header import Header + +# ============== 配置区域 ============== + +# 导入敏感配置 +from config import ( + SMTP_SERVER, SMTP_PORT, SMTP_USER, SMTP_PASSWORD, + FROM_NAME, TO_EMAILS, YOUR_NAME, + DEEPSEEK_API_KEY, DEEPSEEK_API_URL, DEEPSEEK_MODEL, +) + +# 监控的局域网IP列表 +TARGET_IPS = [ + "192.168.2.60", +] + +# 天气位置配置 +WEATHER_LOCATION = "成都" + +# 运行时间配置 +START_HOUR = 2 # 凌晨2点开始 +END_HOUR = 22 # 晚上10点结束 +ALERT_HOUR = 8 # 8点 +ALERT_MINUTE = 15 # 15分 + +# 状态文件目录 +STATE_DIR = "/home/lhr/storage/TaiYangGongGong/.ping_state" + +# 天气和新闻缓存过期时间（秒） +CACHE_EXPIRE = 3600 # 1小时 + +# 天气/新闻缓存文件 +WEATHER_CACHE_FILE = os.path.join(STATE_DIR, "weather_cache.json") +NEWS_CACHE_FILE = os.path.join(STATE_DIR, "news_cache.json") + +DEFAULT_HTTP_HEADERS = { + "User-Agent": "Mozilla/5.0", +} + +# ============== 邮件发送函数 ============== + +def send_email(subject, html_body): + """发送邮件""" + msg = MIMEMultipart('alternative') + msg['From'] = Header(FROM_NAME, 'utf-8').encode() + f" <{SMTP_USER}>" + msg['To'] = ", ".join(TO_EMAILS) + msg['Subject'] = str(Header(subject, 'utf-8')) + + msg.attach(MIMEText(html_body, 'html', 'utf-8')) + + try: + with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server: + server.starttls() + server.login(SMTP_USER, SMTP_PASSWORD) + server.sendmail(SMTP_USER, TO_EMAILS, msg.as_string()) + print(f"[{datetime.datetime.now()}] 邮件发送成功: {subject}") + return True + except Exception as e: + print(f"[{datetime.datetime.now()}] 邮件发送失败: {e}") + return False + +def generate_morning_message(status, arrived_time=None, weather_summary=None, news_summary=None): + """让AI生成自然的早安文案，第一段单独生成不受干扰，第二段衔接第一段，第三段直接用""" + # 第一段：太阳公公状态（AI单独生成，保留关键信息） + sun_status = f"已起床，{arrived_time}就溜出来了" if status == "arrived" else "还在睡觉，还没看到影子" + sun_prompt = f"""请根据太阳公公的起床情况，写一段50字左右的早安问候语： + +太阳公公状态：{sun_status} + +⚠️重要：时间已经是24小时制（如 21:48），直接用这个时间写进文案，不要自己换算或加"上午/下午"！ + +要求： +- 语气俏皮可爱，像朋友聊天 +- 开头加一个和内容相符的emoji +- 50字左右，不要太长 +- 不要用分隔符或标题 +- 只返回这段文字""" + + first_paragraph = ask_deepseek(sun_prompt, system_prompt="你是一个贴心的朋友，写温暖俏皮的早安文案。") + first_paragraph = first_paragraph if first_paragraph else ( + f"🌅 太阳公公终于起床啦，{arrived_time}就偷偷溜出来了~" if status == "arrived" + else "😴 太阳公公这会儿还没醒呢，别急，我帮你盯着着呢~" + ) + + # 第二段：AI根据第一段生成天气段 + weather_prompt = f"""请根据【第一段】的语气，自然过渡到今日天气： + +【第一段】 +{first_paragraph} + +【今日天气】 +{weather_summary if weather_summary else '今天天气信息获取失败'} + +要求： +- 语气要接住【第一段】，像朋友聊天一样自然过渡到天气 +- 50-80字左右 +- 不要用分隔符或标题 +- 不要用列表格式 +- 只返回第二段""" + + second_paragraph = ask_deepseek(weather_prompt, system_prompt="你是一个贴心的朋友，写温暖自然的早安文案。") + second_paragraph = second_paragraph if second_paragraph else f"今日天气：{weather_summary}" + + # 第三段直接用news_summary（已经是串联好的内容） + return f"{first_paragraph}\n\n{second_paragraph}\n\n{news_summary}" + +def build_email_html(status, arrived_time=None, weather_summary=None, news_summary=None): + """构建邮件HTML内容""" + if status == "arrived": + subject = f"😤 早上坏" + else: + subject = f"😊 早上好" + + # 获取AI生成的文案 + morning_message = generate_morning_message(status, arrived_time, weather_summary, news_summary) + + html = f""" + + +
+ {morning_message.replace('\n', '
')} +
+
+

此邮件由系统自动发送

+ + + """ + + return subject, html + +# ============== Ping 检测函数 ============== + +def ping_host(ip, timeout=2): + """检测主机是否在线""" + try: + # Windows 使用 ping -n, 其他使用 ping -c + cmd = ["ping", "-c", "1", "-W", str(timeout), ip] + result = subprocess.run(cmd, capture_output=True, text=True) + return result.returncode == 0 + except Exception: + return False + +# ============== 状态管理 ============== + +def get_today_str(): + return datetime.datetime.now().strftime("%Y-%m-%d") + +def load_state(): + """加载状态文件""" + state_file = os.path.join(STATE_DIR, "monitor_state.json") + if not os.path.exists(state_file): + return {} + + try: + import json + with open(state_file, 'r') as f: + return json.load(f) + except Exception: + return {} + +def save_state(state): + """保存状态文件""" + os.makedirs(STATE_DIR, exist_ok=True) + state_file = os.path.join(STATE_DIR, "monitor_state.json") + + try: + import json + with open(state_file, 'w') as f: + json.dump(state, f, indent=2) + except Exception as e: + print(f"[{datetime.datetime.now()}] 保存状态失败: {e}") + +def should_send_arrived_today(state, ip): + """检查今天是否已发送过到岗通知""" + today = get_today_str() + ip_state = state.get(ip, {}) + return ip_state.get("arrived_date") == today and ip_state.get("arrived_sent", False) + +def should_send_not_arrived_today(state, ip): + """检查今天是否已发送过未到通知""" + today = get_today_str() + ip_state = state.get(ip, {}) + return ip_state.get("not_arrived_date") == today and ip_state.get("not_arrived_sent", False) + +def mark_arrived_sent(state, ip): + """标记已发送到岗通知""" + today = get_today_str() + if ip not in state: + state[ip] = {} + state[ip]["arrived_date"] = today + state[ip]["arrived_sent"] = True + state[ip]["arrived_time"] = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + save_state(state) + +def mark_not_arrived_sent(state, ip): + """标记已发送未到通知""" + today = get_today_str() + if ip not in state: + state[ip] = {} + state[ip]["not_arrived_date"] = today + state[ip]["not_arrived_sent"] = True + save_state(state) + +def reset_daily_state(state): + """重置新一天的标记""" + today = get_today_str() + changed = False + for ip in state: + ip_state = state[ip] + if ip_state.get("arrived_date") != today: + ip_state["arrived_sent"] = False + ip_state["not_arrived_sent"] = False + changed = True + elif ip_state.get("not_arrived_date") != today: + ip_state["not_arrived_sent"] = False + changed = True + return changed + +# ============== 天气获取函数 ============== + +def is_failed_content(content): + """判断内容是否属于失败结果""" + return not content or "获取失败" in content + +def fetch_text(url, headers=None, timeout=10): + """发起HTTP请求并返回文本内容""" + request_headers = dict(DEFAULT_HTTP_HEADERS) + if headers: + request_headers.update(headers) + + req = urllib.request.Request(url, headers=request_headers) + with urllib.request.urlopen(req, timeout=timeout) as response: + return response.read().decode('utf-8', errors='ignore') + +def get_weather(): + """获取当天天气信息（带缓存）""" + today = get_today_str() + + # 检查缓存 + if os.path.exists(WEATHER_CACHE_FILE): + try: + with open(WEATHER_CACHE_FILE, 'r') as f: + cache = json.load(f) + cached_weather = cache.get("weather", "") + if ( + cache.get("date") == today + and cache.get("location") == WEATHER_LOCATION + and time.time() - cache.get("timestamp", 0) < CACHE_EXPIRE + and not is_failed_content(cached_weather) + ): + return cached_weather + except Exception: + pass + + # 天气信息来源 - 使用wttr.in（免费无需API Key） + weather_info = "" + try: + encoded_location = urllib.parse.quote(WEATHER_LOCATION) + url = f"https://wttr.in/{encoded_location}?format=%l:+%C,+%t,+%h,+feeling:+%f,+%p" + weather_info = fetch_text(url) + if not weather_info.strip(): + weather_info = "天气信息获取失败" + except Exception as e: + print(f"[{datetime.datetime.now()}] 获取天气失败: {e}") + weather_info = "天气信息获取失败" + + # 保存缓存 + os.makedirs(STATE_DIR, exist_ok=True) + cache_data = { + "date": today, + "location": WEATHER_LOCATION, + "timestamp": time.time(), + "weather": weather_info + } + try: + with open(WEATHER_CACHE_FILE, 'w') as f: + json.dump(cache_data, f) + except Exception: + pass + + return weather_info + +# ============== 新闻获取函数 ============== + +def get_hot_news(): + """获取当天热搜新闻（带缓存）""" + today = get_today_str() + + # 检查缓存 + if os.path.exists(NEWS_CACHE_FILE): + try: + with open(NEWS_CACHE_FILE, 'r') as f: + cache = json.load(f) + cached_news = cache.get("news", "") + if ( + cache.get("date") == today + and time.time() - cache.get("timestamp", 0) < CACHE_EXPIRE + and not is_failed_content(cached_news) + ): + return cached_news + except Exception: + pass + + # 从百度热搜页提取热榜关键词 + news_info = "" + try: + html = fetch_text("https://top.baidu.com/board?tab=realtime") + raw_titles = re.findall(r'"word":"(.*?)"', html) + news_items = [] + seen_titles = set() + + for raw_title in raw_titles: + title = raw_title.strip() + if not title or title in seen_titles: + continue + seen_titles.add(title) + news_items.append(title) + if len(news_items) >= 10: + break + + if not news_items: + raise ValueError("未能从百度热搜页面提取标题") + + news_info = "\n".join( + f"{index}. {title}" for index, title in enumerate(news_items, 1) + ) + except Exception as e: + print(f"[{datetime.datetime.now()}] 获取新闻失败: {e}") + + if not news_info: + news_info = "新闻信息获取失败" + + # 保存缓存 + os.makedirs(STATE_DIR, exist_ok=True) + cache_data = { + "date": today, + "timestamp": time.time(), + "news": news_info + } + try: + with open(NEWS_CACHE_FILE, 'w') as f: + json.dump(cache_data, f) + except Exception: + pass + + return news_info + +# ============== DeepSeek AI 处理函数 ============== + +def ask_deepseek(prompt, system_prompt="你是一个有用的助手，用简洁温暖的语言回复。"): + """调用DeepSeek API获取回复""" + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {DEEPSEEK_API_KEY}" + } + + payload = { + "model": DEEPSEEK_MODEL, + "messages": [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": prompt} + ], + "temperature": 0.7, + "max_tokens": 500 + } + + try: + req = urllib.request.Request( + DEEPSEEK_API_URL, + data=json.dumps(payload).encode('utf-8'), + headers=headers, + method='POST' + ) + with urllib.request.urlopen(req, timeout=30) as response: + result = json.loads(response.read().decode('utf-8')) + return result.get("choices", [{}])[0].get("message", {}).get("content", "") + except Exception as e: + print(f"[{datetime.datetime.now()}] DeepSeek API调用失败: {e}") + return None + +def get_weather_summary(weather_info): + """让DeepSeek总结天气信息""" + if not weather_info or "获取失败" in weather_info: + return weather_info + + prompt = f"""请用简洁温暖的语言总结以下天气信息，控制在50字以内，适合放在早安邮件里： + +{weather_info} + +请直接返回总结内容，不要多余的话。""" + + summary = ask_deepseek(prompt, system_prompt="你是一个贴心的早安助手，用温暖简洁的语言回复。") + return summary if summary else weather_info + +def get_news_summary(news_info): + """让DeepSeek总结新闻要点，找出关联，串联成自然的段落""" + if not news_info or "获取失败" in news_info: + return news_info + + prompt = f"""请阅读以下今日热搜新闻，试着找出它们之间的关联或共同话题，然后写成一段200字左右的自然段落，像朋友聊天一样分享这些新闻： + +{news_info} + +要求： +- 找出新闻之间的联系，用自己的话串起来讲，不要念稿 +- 讲的时候像跟朋友吐槽聊天那样，说到哪个新闻就直接说，别用引号包着 +- 语气像在说相声或脱口秀，自然调侃，不要端着 +- 不要用列表或条目，要连贯的段落 +- 字数200字左右 +- 只返回这段新闻解读，不要多余的话""" + + summary = ask_deepseek(prompt, system_prompt="你是一个贴心的早安助手，用温暖自然的语言回复。") + return summary if summary else news_info + +# ============== 时间判断 ============== + +def is_in_monitor_window(): + """检查是否在监控时间窗口内""" + now = datetime.datetime.now() + current_hour = now.hour + + # 凌晨2点到晚上10点 + if current_hour >= START_HOUR and current_hour < END_HOUR: + return True + return False + +def is_alert_time(): + """检查是否到了8:15提醒时间""" + now = datetime.datetime.now() + if now.hour == ALERT_HOUR and now.minute == ALERT_MINUTE: + return True + return False + +# ============== 主监控循环 ============== + +# 缓存天气和新闻摘要 +_cached_weather_summary = None +_cached_news_summary = None + +def get_cached_content(): + """获取缓存的天气和新闻摘要""" + global _cached_weather_summary, _cached_news_summary + + # 如果没有缓存，先获取 + if _cached_weather_summary is None: + print(f"[{datetime.datetime.now()}] 正在获取天气信息...") + weather_info = get_weather() + _cached_weather_summary = get_weather_summary(weather_info) + print(f"[{datetime.datetime.now()}] 天气摘要: {_cached_weather_summary[:50]}...") + + if _cached_news_summary is None: + print(f"[{datetime.datetime.now()}] 正在获取新闻信息...") + news_info = get_hot_news() + _cached_news_summary = get_news_summary(news_info) + print(f"[{datetime.datetime.now()}] 新闻摘要: {_cached_news_summary[:50]}...") + + return _cached_weather_summary, _cached_news_summary + +def reset_daily_content_cache(): + """重置每日内容缓存""" + global _cached_weather_summary, _cached_news_summary + _cached_weather_summary = None + _cached_news_summary = None + +def monitor_once(state): + """执行一次监控检查""" + weather_summary, news_summary = get_cached_content() + + for ip in TARGET_IPS: + online = ping_host(ip) + + if online: + print(f"[{datetime.datetime.now()}] {ip} 在线") + + if not should_send_arrived_today(state, ip): + arrived_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + subject, html = build_email_html("arrived", arrived_time, weather_summary, news_summary) + if send_email(subject, html): + mark_arrived_sent(state, ip) + else: + print(f"[{datetime.datetime.now()}] {ip} 离线") + +def check_not_arrived_alert(state): + """检查是否需要发送未到提醒""" + weather_summary, news_summary = get_cached_content() + + for ip in TARGET_IPS: + # 只有在目标仍然离线的情况下才发送未到提醒 + if not ping_host(ip) and not should_send_not_arrived_today(state, ip): + subject, html = build_email_html("not_arrived", None, weather_summary, news_summary) + if send_email(subject, html): + mark_not_arrived_sent(state, ip) + +def main(): + print(f"[{datetime.datetime.now()}] LAN监控脚本启动") + print(f"监控IP: {TARGET_IPS}") + print(f"运行时间: {START_HOUR}:00 - {END_HOUR}:00") + print(f"提醒时间: {ALERT_HOUR}:{ALERT_MINUTE:02d}") + print(f"DeepSeek API: {DEEPSEEK_API_URL}") + print("-" * 50) + + os.makedirs(STATE_DIR, exist_ok=True) + + last_alert_day = None + last_content_day = None + + while True: + now = datetime.datetime.now() + current_day = now.strftime("%Y-%m-%d") + + # 加载状态 + state = load_state() + + # 检查是否需要重置每日标记 + if reset_daily_state(state): + save_state(state) + print(f"[{now}] 新的一天，重置通知状态") + + # 检查是否需要重置内容缓存 + if last_content_day != current_day: + reset_daily_content_cache() + last_content_day = current_day + print(f"[{now}] 新的一天，重置天气新闻缓存") + + # 检查是否在监控时间窗口内 + if is_in_monitor_window(): + # 执行一次监控 + monitor_once(state) + + # 检查是否到8:15提醒时间（每小时检查一次） + if is_alert_time() and last_alert_day != current_day: + check_not_arrived_alert(state) + last_alert_day = current_day + else: + print(f"[{now}] 不在监控时间窗口 ({START_HOUR}:00 - {END_HOUR}:00)，等待中...") + + # 每30秒检查一次 + time.sleep(30) + +if __name__ == "__main__": + main() diff --git a/test_ping.py b/test_ping.py new file mode 100644 index 0000000..adc4ae8 --- /dev/null +++ b/test_ping.py @@ -0,0 +1,79 @@ +#!/usr/bin/env python3 +"""测试脚本 - 运行10秒扫描并发送邮件""" + +import os +import sys +import time +import datetime +import subprocess + +# 导入主脚本的配置和函数 +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import ping_monitor + +def ping_host(ip, timeout=2): + """检测主机是否在线""" + try: + cmd = ["ping", "-c", "1", "-W", str(timeout), ip] + result = subprocess.run(cmd, capture_output=True, text=True) + return result.returncode == 0 + except Exception: + return False + +def main(): + print(f"[{datetime.datetime.now()}] 测试脚本启动") + print(f"监控IP: {ping_monitor.TARGET_IPS}") + print(f"天气位置: {ping_monitor.WEATHER_LOCATION}") + print("-" * 50) + + # 确保状态目录存在 + os.makedirs(ping_monitor.STATE_DIR, exist_ok=True) + + # 获取天气和新闻 + print("正在获取天气和新闻...") + weather_info = ping_monitor.get_weather() + news_info = ping_monitor.get_hot_news() + print(f"天气原始信息: {weather_info[:100]}...") + print(f"新闻原始信息: {news_info[:100]}...") + + weather_summary = ping_monitor.get_weather_summary(weather_info) + news_summary = ping_monitor.get_news_summary(news_info) + print(f"天气摘要: {weather_summary}") + print(f"新闻摘要: {news_summary}") + print("-" * 50) + + # 测试Ping所有IP + online_ips = [] + offline_ips = [] + for ip in ping_monitor.TARGET_IPS: + online = ping_host(ip) + if online: + online_ips.append(ip) + print(f"[{datetime.datetime.now()}] {ip} 在线") + else: + offline_ips.append(ip) + print(f"[{datetime.datetime.now()}] {ip} 离线") + + print("-" * 50) + + # 加载状态 + state = ping_monitor.load_state() + arrived_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + # 如果有在线的IP，发送到岗邮件 + if online_ips: + print(f"发送在线邮件 for: {online_ips}") + subject, html = ping_monitor.build_email_html("arrived", arrived_time, weather_summary, news_summary) + ping_monitor.send_email(subject, html) + + # 如果有离线的IP，发送未到邮件 + if offline_ips: + print(f"发送离线邮件 for: {offline_ips}") + subject, html = ping_monitor.build_email_html("not_arrived", None, weather_summary, news_summary) + ping_monitor.send_email(subject, html) + + print("-" * 50) + print(f"[{datetime.datetime.now()}] 测试完成") + +if __name__ == "__main__": + main()