import requests from bs4 import BeautifulSoup import smtplib from email.mime.text import MIMEText import time import random import logging from datetime import datetime # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[logging.FileHandler('monitor.log'), logging.StreamHandler()] ) # 邮件配置 (请替换为您的实际信息) SMTP_SERVER = 'smtp.qq.com' SMTP_PORT = 587 SMTP_USER = '[email protected]' SMTP_PASSWORD = 'lfnnoqqokcozbfha' TO_EMAIL = '[email protected]' # 商品链接列表 (添加您要监控的商品) PRODUCT_URLS = [ 'https://2.taobao.com/item.htm?id=123456789', 'https://2.taobao.com/item.htm?id=987654321', 'https://2.taobao.com/item.htm?id=567891234' ] # 存储商品状态历史 product_status = {url: None for url in PRODUCT_URLS} def send_email(subject, body): """发送邮件通知""" try: msg = MIMEText(body, 'plain', 'utf-8') msg['Subject'] = subject msg['From'] = SMTP_USER msg['To'] = TO_EMAIL with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server: server.starttls() server.login(SMTP_USER, SMTP_PASSWORD) server.send_message(msg) logging.info(f"邮件发送成功: {subject}") return True except Exception as e: logging.error(f"邮件发送失败: {str(e)}") return False def get_product_status(url): """获取商品状态""" try: headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', 'Accept-Language': 'zh-CN,zh;q=0.9' } response = requests.get(url, headers=headers, timeout=10) response.raise_for_status() soup = BeautifulSoup(response.text, 'html.parser') # 解析商品状态 - 根据咸鱼实际页面结构调整 # 示例1: 检查商品是否下架 offline_element = soup.find('div', class_='offline') if offline_element and '已下架' in offline_element.text: return '已下架' # 示例2: 检查商品是否售出 sold_element = soup.find('div', class_='sold-out') if sold_element and '已售出' in sold_element.text: return '已售出' # 示例3: 检查价格区域是否存在 price_element = soup.find('span', class_='price') if not price_element: return '状态异常' return '在售中' except requests.exceptions.RequestException as e: logging.error(f"请求错误 {url}: {str(e)}") return '请求失败' except Exception as e: logging.error(f"解析错误 {url}: {str(e)}") return '解析失败' def monitor_products(): """监控多个商品状态""" while True: for url in PRODUCT_URLS: try: current_status = get_product_status(url) previous_status = product_status[url] # 记录状态变化 if current_status != previous_status: timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") logging.info(f"状态变化: {url} | 旧状态: {previous_status} | 新状态: {current_status}") # 发送通知 if current_status in ['已下架', '已售出', '状态异常']: subject = f"咸鱼商品状态变化: {current_status}" body = f"商品链接: {url}\n状态: {current_status}\n时间: {timestamp}" send_email(subject, body) # 更新状态 product_status[url] = current_status except Exception as e: logging.error(f"监控错误 {url}: {str(e)}") # 随机延迟 (0.5-5秒) delay = random.uniform(0.5, 5) logging.info(f"等待 {delay:.2f} 秒后继续监控...") time.sleep(delay) if __name__ == "__main__": logging.info("咸鱼商品监控脚本启动") try: monitor_products() except KeyboardInterrupt: logging.info("监控脚本已手动停止") except Exception as e: logging.error(f"脚本异常终止: {str(e)}") 以上代码启动后只是闪一下运行框，请帮我完善一下

<think>我们有两个主要问题需要解决： 1. Python脚本运行时命令行窗口一闪而过的问题（通常发生在Windows系统中双击运行脚本时） 2. 完善商品监控脚本的日志和异常处理 首先，解决命令行窗口一闪而过的问题： 原因：当双击运行Python脚本时，系统会打开一个命令行窗口来执行脚本，脚本执行完毕后窗口会自动关闭。如果脚本中有错误导致立即退出，或者脚本正常执行完毕，都会导致窗口关闭。 解决方案： 方案1：在脚本末尾添加一个等待用户输入的命令，这样窗口会保持打开直到用户按下回车。 方案2：通过命令行运行脚本（打开cmd，然后执行python 脚本名.py）。 方案3：使用try-except捕获异常，并在异常发生时暂停。 我们可以在脚本的末尾添加以下代码，这样当脚本运行结束时（无论是正常结束还是异常结束）都会暂停： ```python input("按Enter键退出...") ``` 但是，我们需要注意的是，我们的监控脚本是一个无限循环（monitor_products函数中是while True），所以通常不会自己结束。因此，窗口一闪而过的问题可能是由于脚本启动时发生异常导致立即退出。 因此，我们需要完善脚本的异常处理，确保在发生异常时能够提供足够的错误信息并暂停。 其次，完善商品监控脚本的日志和异常处理： 我们已经配置了日志记录，但可以进一步细化异常捕获，特别是在主函数中。 修改后的主函数部分： ```python if __name__ == "__main__": logging.info("咸鱼商品监控脚本启动") try: monitor_products() except KeyboardInterrupt: logging.info("监控脚本已手动停止") # 这里可以添加发送通知的功能，告知脚本被手动停止 except Exception as e: logging.exception(f"脚本异常终止: {str(e)}") # 在发生未处理异常时暂停，以便查看错误信息 input("按Enter键退出...") # 也可以选择发送邮件通知异常 send_email("咸鱼监控脚本异常终止", f"错误信息: {str(e)}") finally: logging.info("脚本退出") ``` 这样修改后，当脚本遇到未捕获的异常时，会记录异常详细信息，并暂停等待用户按Enter键，同时发送邮件通知。 另外，我们还可以在监控循环中增加更细粒度的异常处理，确保一个商品的监控失败不会影响其他商品的监控。 我们已经做了： ```python for url in PRODUCT_URLS: try: # 监控单个商品 except Exception as e: logging.error(f"监控错误 {url}: {str(e)}") ``` 这样即使某个商品监控出错，也会继续监控下一个商品。 现在，我们整合这些改进到脚本中。 修改后的完整脚本（主要针对异常处理和退出机制）：</think># 解决Python脚本窗口一闪而过问题及完善监控脚本 ## 一、解决命令行窗口一闪而过的问题 当双击运行Python脚本时出现窗口一闪而过的情况，通常是由于脚本执行完毕或遇到错误立即退出造成的。以下是几种解决方案： ### 方法1：在脚本末尾添加输入等待 ```python if __name__ == "__main__": try: # 主程序逻辑 except Exception as e: import traceback traceback.print_exc() finally: input("按Enter键退出...") # 添加此行 ``` ### 方法2：通过命令行运行脚本 1. 打开命令提示符（Win+R 输入 `cmd`） 2. 导航到脚本目录：`cd 你的脚本目录` 3. 运行脚本：`python 你的脚本.py` 4. 执行结果和错误信息将保留在命令窗口中 ### 方法3：使用批处理文件 创建 `.bat` 文件： ```batch @echo off python "你的脚本.py" pause ``` ## 二、完善商品监控脚本的日志和异常处理 ### 1. 增强日志配置 ```python import logging from logging.handlers import RotatingFileHandler # 配置日志 logger = logging.getLogger('xianyu_monitor') logger.setLevel(logging.DEBUG) # 文件日志 - 按大小轮转 file_handler = RotatingFileHandler( 'monitor.log', maxBytes=5*1024*1024, # 5MB backupCount=3 ) file_handler.setFormatter(logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s' )) # 控制台日志 console_handler = logging.StreamHandler() console_handler.setFormatter(logging.Formatter( '%(asctime)s - %(levelname)s - %(message)s' )) logger.addHandler(file_handler) logger.addHandler(console_handler) ``` ### 2. 完善异常处理框架 ```python def get_product_status(url): try: # 原有逻辑... except requests.exceptions.RequestException as e: logger.error(f"请求异常 [{url}]: {type(e).__name__} - {str(e)}") return '请求失败' except Exception as e: logger.exception(f"解析异常 [{url}]:") # 记录完整堆栈 return '解析失败' def monitor_products(): while True: try: # 监控逻辑... except KeyboardInterrupt: logger.info("用户中断监控") raise except Exception as e: logger.critical(f"监控主循环异常: {type(e).__name__} - {str(e)}") # 指数退避重试 delay = min(300, 2 ** retry_count) # 最大5分钟 logger.info(f"{delay}秒后重试...") time.sleep(delay) retry_count += 1 ``` ### 3. 添加监控指标统计 ```python class MonitorStats: def __init__(self): self.start_time = time.time() self.total_checks = 0 self.success_checks = 0 self.status_changes = 0 def log_check(self, success=True): self.total_checks += 1 if success: self.success_checks += 1 def log_change(self): self.status_changes += 1 def get_uptime(self): return time.time() - self.start_time def get_success_rate(self): if self.total_checks == 0: return 0.0 return self.success_checks / self.total_checks * 100 # 在监控循环中使用 stats = MonitorStats() for url in PRODUCT_URLS: # ... if current_status != previous_status: stats.log_change() stats.log_check(current_status not in ['请求失败', '解析失败']) ``` ### 4. 添加定时状态报告 ```python def send_status_report(stats): """发送定时状态报告""" subject = "咸鱼监控状态报告" body = ( f"运行时间: {stats.get_uptime()/3600:.2f} 小时\n" f"总检查次数: {stats.total_checks}\n" f"成功率: {stats.get_success_rate():.2f}%\n" f"状态变化次数: {stats.status_changes}" ) send_email(subject, body) # 在监控循环中添加 if time.time() - last_report_time > 3600: # 每小时报告 send_status_report(stats) last_report_time = time.time() ``` ### 5. 完整异常处理框架 ```python if __name__ == "__main__": try: logger.info("咸鱼商品监控脚本启动") stats = MonitorStats() last_report_time = time.time() monitor_products() except KeyboardInterrupt: logger.info("监控脚本已手动停止") send_email("监控停止通知", "用户手动停止了咸鱼监控脚本") except Exception as e: logger.exception("脚本致命错误:") send_email( "监控脚本崩溃通知", f"脚本遇到未处理异常:\n{type(e).__name__}: {str(e)}" ) finally: logger.info("脚本退出") # 仅当直接运行时才等待输入 if not sys.argv[0].endswith('.py'): # 判断是否双击运行 input("按Enter键退出...") ``` ## 三、调试技巧 ### 1. 使用IDE调试 推荐使用VSCode或PyCharm： 1. 设置断点 2. 启动调试模式 3. 使用变量监视器 4. 使用步进执行(Step Into/Over) ### 2. 日志级别动态调整 ```python import argparse parser = argparse.ArgumentParser() parser.add_argument('--debug', action='store_true', help='启用调试模式') args = parser.parse_args() if args.debug: logger.setLevel(logging.DEBUG) logger.debug("调试模式已启用") ``` ### 3. 关键检查点断言 ```python def get_product_status(url): # ... response = requests.get(url, headers=headers, timeout=10) assert response.status_code == 200, f"非200状态码: {response.status_code}" # ... ``` ### 4. 保存错误页面快照 ```python except Exception as e: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"error_{timestamp}.html" with open(filename, 'w', encoding='utf-8') as f: f.write(response.text) logger.error(f"保存错误页面到: {filename}") ``` ## 四、部署建议 1. **Windows任务计划**： - 创建批处理文件：`start_monitor.bat` ```batch @echo off python "C:\path\to\your\script.py" >> "C:\path\to\log.txt" 2>&1 ``` - 使用任务计划程序定期运行 2. **Linux系统服务**： ```bash # 创建服务文件 /etc/systemd/system/xianyu-monitor.service [Unit] Description=Xianyu Product Monitor [Service] ExecStart=/usr/bin/python3 /path/to/script.py Restart=always RestartSec=3 [Install] WantedBy=multi-user.target ``` 3. **Docker容器化**： ```Dockerfile FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD ["python", "monitor.py"] ``` ## 五、监控脚本优化建议 1. **添加代理支持**： ```python proxies = { 'http': 'http://user:pass@proxy_ip:port', 'https': 'https://user:pass@proxy_ip:port' } response = requests.get(url, proxies=proxies, timeout=10) ``` 2. **实现状态持久化**： ```python import json def save_status(): with open('status.json', 'w') as f: json.dump(product_status, f) def load_status(): try: with open('status.json', 'r') as f: return json.load(f) except FileNotFoundError: return {url: None for url in PRODUCT_URLS} ``` 3. **添加健康检查端点**： ```python from flask import Flask app = Flask(__name__) @app.route('/health') def health(): return {'status': 'running', 'uptime': stats.get_uptime()} # 在单独线程运行 import threading threading.Thread(target=lambda: app.run(port=5000), daemon=True).start() ``` ## 相关数学表达 监控脚本的可靠性可以通过成功率度量： $$ \text{成功率} = \frac{\text{成功请求数}}{\text{总请求数}} \times 100\% $$ 重试机制采用指数退避算法： $$ \text{重试延迟} = \min(\text{最大延迟}, 2^{\text{重试次数}}) $$ ##