from playwright.sync_api import sync_playwright import sqlite3 import smtplib from email.mime.text import MIMEText from email.header import Header import random import logging import time from logging.handlers import TimedRotatingFileHandler # # 创建一个选品logger # logger = logging.getLogger('detailLogger') # logger.setLevel(logging.DEBUG) # 设置日志级别 # # 创建一个handler，用于写入文件，并根据时间自动轮换日志文件 # detail_handler = TimedRotatingFileHandler('日志log/日志.txt', when='midnight', interval=1, backupCount=30,encoding='utf-8') # detail_handler.setFormatter(logging.Formatter('%(asctime)s - %(message)s')) # # 添加handler到logger # logger.addHandler(detail_handler) # 配置日志记录器 logger_cmd1 = logging.getLogger('logger_cmd1') logger_cmd1.setLevel(logging.DEBUG) # 配置控制台处理器和格式器 console_handler = logging.StreamHandler() console_handler.setLevel(logging.DEBUG) formatter = logging.Formatter('[%(asctime)s] %(message)s') console_handler.setFormatter(formatter) # 将处理器添加到记录器中 logger_cmd1.addHandler(console_handler) def run(session='', sender_email='', receiver_email='', smtp_proxy='', smtp_account='', smtp_password='', email_alias='', error_count=3,you_scan_interval=30,logger=''): # 初始化数据 count = 0 try: with sync_playwright() as playwright: browser = playwright.chromium.launch(headless=True) context = browser.new_context() page = context.new_page() # 添加session cookie cookies = [ { 'name': 'session', 'value': session, # 替换为实际的session值 'domain': 'buff.163.com', 'path': '/', } ] context.add_cookies(cookies) # 商品池子 # goods_info = [776831, 911801, 42234, 33887, 781577, 968110, 968048, 968047, 968046, 968045, 968044, 968043, 968042, 968041, 968040, 968039, 968038, 968037, 968036, 968035, 968034, 968033, 968032, 968031, 968030, 968029, 968028, 968027, 968026, 968025, 968024, 968023, 968022, 968021, 968020, 968019, 968018, ] goods_info = [968020] # 从goods_info随机获取一个值 random_number = random.choice(goods_info) # Step 1: Visit the initial URL page.goto(f"https://buff.163.com/goods/{random_number}?from=market#tab=selling") # Step 2: Click the "无法扫描二维码" button # 等待page加载完毕 page.wait_for_load_state('networkidle') page.wait_for_function('''() => { return document.readyState === 'complete' }''') # logger.info(f"点击无法扫描二维码按钮-是否存在(0:不存在，1:存在): {page.locator('span.yidun_smsbox-text--manual').count()}") if page.locator('span.yidun_smsbox-text--manual').count() <= 0 : logger.info("清空表sms_codes") conn = sqlite3.connect('settings.db') cursor = conn.cursor() cursor.execute('''CREATE TABLE IF NOT EXISTS sms_codes (edit_content TEXT, send_content_short TEXT)''') cursor.execute('delete from sms_codes') conn.commit() conn.close() # Close the browser context.close() browser.close() logger.info("未监测到短信数据") return 'success' page.click('span.yidun_smsbox-text--manual') # Step 3: Wait for navigation to the new page page.wait_for_load_state('networkidle') page.wait_for_function('''() => { return document.readyState === 'complete' }''') # Step 4: Extract the required content edit_content = page.inner_text('span.yidun_smsbox-manual--edit-content') send_content_short = page.inner_text('span.yidun_smsbox-manual--send-content__short') # Step 5: Save to SQLite database conn = sqlite3.connect('settings.db') cursor = conn.cursor() cursor.execute('''CREATE TABLE IF NOT EXISTS sms_codes (edit_content TEXT, send_content_short TEXT)''') cursor.execute('''INSERT INTO sms_codes (edit_content, send_content_short) VALUES (?, ?)''', (edit_content, send_content_short)) conn.commit() # 判断sms_codes表中的数据量是否大于3 cursor.execute('''SELECT COUNT(*) FROM sms_codes''') # 获取执行结果 count = cursor.fetchone()[0] conn.close() logger.info(f'count >= int(error_count) + 1 ：{count >= int(error_count) + 1}') if count >= int(error_count) + 1: logger.info(f'达到异常关闭条件,连续异常次数:{count}，指标为:{error_count}，返回True，结束程序') return 'stop' else: pass logger_cmd1.info(f'需要短信验证，次数{int(count)}！') logger_cmd1.info(f'发送{edit_content}到{send_content_short}，等待{float(you_scan_interval)}秒！') logger.info(f'短信发送到: {send_content_short}，短信内容: {edit_content}') # 发件人邮箱账号 sender = sender_email # 收件人邮箱账号，可以是多个 receivers = receiver_email # 邮件正文内容 message = MIMEText(f'编辑短信{edit_content}', 'plain', 'utf-8') # 邮件主题 message['Subject'] = Header(f'{email_alias}编辑短信{edit_content}', 'utf-8') # 发件人和收件人 message['From'] = sender_email message['To'] = receiver_email try: # 初始化 smtpObj 为 None smtpObj = None # 连接到SMTP服务器，使用SSL连接（端口465）或TLS连接（端口587） smtpObj = smtplib.SMTP_SSL(smtp_proxy, 465) # 使用SSL连接示例 # smtpObj = smtplib.SMTP("smtp.126.com", 587) # 使用TLS连接示例，需要先调用smtpObj.starttls() # smtpObj.starttls() # 如果使用TLS连接，取消注释这行并注释上一行中的SSL连接部分 # 登录SMTP服务器 logger.info("SMTP sender:{sender} smtp_password:{smtp_password}") smtpObj.login(sender, smtp_password) # 注意：出于安全考虑，建议使用应用专用密码而非你的邮箱密码 logger.info("SMTP 登录成功,sender:{sender} smtp_password:{smtp_password}") # 发送邮件 smtpObj.sendmail(sender, receivers, message.as_string()) logger.info("邮件发送成功") except smtplib.SMTPAuthenticationError as e: logger.error(f"SMTP 认证失败: {e}") except smtplib.SMTPException as e: logger.info("Error: 无法发送邮件", e) finally: # 检查 smtpObj 是否被成功创建 if smtpObj is not None: smtpObj.quit() # wait_times = 0 # 监测span.yidun_smsbox-manual--edit-content元素，如果元素消失了才能关闭浏览器,超过了3分钟自动关闭 # while page.locator('span.yidun_smsbox-manual--edit-content').count() > 0 and wait_times < 180: # logger.info(f"等待系统验证通过中........,已等待{wait_times}秒") # wait_times += 1 # time.sleep(1) # if page.locator('span.yidun_smsbox-manual--edit-content').count() <= 0: # logger.info(f"系统验证通过") # if wait_times >= 180: # logger.info(f"等待系统验证通过超时") logger.info(f'等待短信验证，等待{float(you_scan_interval)}秒') time.sleep(float(you_scan_interval)) # Close the browser context.close() browser.close() logger.info(f'未达到异常关闭条件，,连续异常次数:{count}，指标为:{error_count} 返回 error not stop') return 'error not stop' except Exception as e: logger.info(f"未监测到短信数据: {e}") return 'error not stop1' finally: pass # 如果count大于等于三，则返回'stop' # if count >= int(error_count): # logger.info(f'达到异常关闭条件,连续异常次数:{count}，指标为:{error_count}，返回True，结束程序') # return True # else: # # logger.info(f'未达到异常关闭条件，,连续异常次数:{count}，指标为:{error_count} 返回False') # return False if __name__ == "__main__": result = run( session='session', sender_email='sender_email', receiver_email='receiver_email', smtp_proxy='smtp_proxy', smtp_account='smtp_account', smtp_password='smtp_password', email_alias='email_alias', error_count='3', you_scan_interval='20' ) # logger.info(result)