#!/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()