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 import requests as req import re import json logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(message)s') logger = logging.getLogger(__name__) def run(session='', sender_email='', receiver_email='', smtp_proxy='', smtp_account='', smtp_password='', email_alias='', error_count=3): # 初始化数据 count = 0 dx_my_headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.131 Safari/537.36", "X-Requested-With": "XMLHttpRequest" } try: # get 请求 dx_my_headers.update({ 'Cookie': f'sessions={session}' }) api_url="https://c.dun.163.com/api/v3/get?referer=https%3A%2F%2Fbuff.163.com%2Fgoods%2F921509&zoneId=CN31&dt=8gRTZRf8UUxEA1EREFPCNl0v%2BDhw0mZg&acToken=9ca17ae2e6ffcda170e2e6eeaecf6088eae189c63baeb88fb7d14a938f8f87db43af92f783cf52f7a8bab6f02af0feaec3b92af4afb698e15491bdb7d2eb4f878a9ab3d84e8d99add3e55295adf984d94a9288ee9e&id=176df3efe81a4945978a3dc4ce277139&fp=d1%5CgemagKNZrEyaLbQJldfuJj39Z963luTbM%2BUHtRgoypfCmVHuL2aiSo84O2mS63vHmfLXXrsTstfCUWX1y2W4Pj22UXQwe7zo8w%2FQZ805%2FWiUzxm%5C%5CANYurvHNf%2BDU%2Fkh9Ka2031QT6%2BczlNCUHJlERqoDKZHtpjwqOPjcpiIC7DNK%3A1740976364913&https=true&type=undefined&version=2.27.2&dpr=2&dev=1&cb=65e.pcSwFeSYbuuunMPYWtbkxsAhWWAkAPWzHEMpZ.aCjclRQDjRJA3A.r2mBLsH2uCaiubbhQaXbkux2KMiENC.o%2FY7&ipv6=false&runEnv=10&group=&scene=&lang=zh-CN&sdkVersion=undefined&iv=4&width=320&audio=false&sizeType=10&smsVersion=v3&token=&callback=__JSONP_p2z3mwq_0" api2_res = req.get(api_url, headers=dx_my_headers) # 获取api2_res的body内容 logger.info(f'接口返回：{api2_res.text}') # 使用正则表达式提取 JSON 部分 json_string = re.search(r'\((.*)\)', api2_res.text).group(1) logger.info(json_string) # 将 JSON 字符串转换为 Python 字典 data = json.loads(json_string) error_value = data['error'] # 成功的话 if error_value == '0': msg_content = data['data']['front'] msg_content_list = msg_content.split(',') edit_content = msg_content_list[0] send_content_short = msg_content_list[1] logger.info(f'短信内容: {edit_content},短信发送到: {send_content_short}') # 发件人邮箱账号 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: # 连接到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服务器 smtpObj.login(sender, smtp_password) # 注意：出于安全考虑，建议使用应用专用密码而非你的邮箱密码 # 发送邮件 smtpObj.sendmail(sender, receivers, message.as_string()) logger.info("邮件发送成功") except smtplib.SMTPException as e: logger.info("Error: 无法发送邮件", e) finally: smtpObj.quit() # 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("暂停180秒，给短信验证预留时间") time.sleep(180) else: # 失败的话 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() # logger.info("未监测到短信数据") return 'success' # with sync_playwright() as playwright: # browser = playwright.chromium.launch(headless=False) # 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') # # 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 False # page.click('span.yidun_smsbox-text--manual') # # # Step 3: Wait for navigation to the new page # page.wait_for_load_state('networkidle') # # 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') # 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: # # 连接到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服务器 # smtpObj.login(sender, smtp_password) # 注意：出于安全考虑，建议使用应用专用密码而非你的邮箱密码 # # 发送邮件 # smtpObj.sendmail(sender, receivers, message.as_string()) # logger.info("邮件发送成功") # except smtplib.SMTPException as e: # logger.info("Error: 无法发送邮件", e) # finally: # smtpObj.quit() # # # 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() # # 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"等待系统验证通过超时") # # # Close the browser # context.close() # browser.close() # 如果count大于等于三，则返回'stop' if count >= int(error_count): logger.info(f'达到异常关闭条件,连续异常次数:{count}，指标为:{error_count}，返回True，结束程序') return 'stop' else: logger.info(f'未达到异常关闭条件，,连续异常次数:{count}，指标为:{error_count} 返回False') return 'error not stop' except Exception as e: logger.info(f"未监测到短信数据: {e}") return 'error not stop' finally: pass 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' ) logger.info(result)