diff --git a/sau_backend/myUtils/login.py b/sau_backend/myUtils/login.py new file mode 100644 index 0000000..738a3ea --- /dev/null +++ b/sau_backend/myUtils/login.py @@ -0,0 +1,222 @@ +import asyncio +import sqlite3 + +from playwright.async_api import async_playwright + +from sau_backend.myUtils.auth import check_cookie +from sau_backend.utils.base_social_media import set_init_script +import uuid +from pathlib import Path +from sau_backend.conf import BASE_DIR + +# 抖音登录 +async def douyin_cookie_gen(id,status_queue): + url_changed_event = asyncio.Event() + async def on_url_change(): + # 检查是否是主框架的变化 + if page.url != original_url: + url_changed_event.set() + async with async_playwright() as playwright: + options = { + 'headless': False + } + # Make sure to run headed. + browser = await playwright.chromium.launch(**options) + # Setup context however you like. + context = await browser.new_context() # Pass any options + context = await set_init_script(context) + # Pause the page, and start recording manually. + page = await context.new_page() + await page.goto("https://creator.douyin.com/") + original_url = page.url + img_locator = page.get_by_role("img", name="二维码") + # 获取 src 属性值 + src = await img_locator.get_attribute("src") + print("✅ 图片地址:", src) + status_queue.put(src) + # 监听页面的 'framenavigated' 事件,只关注主框架的变化 + page.on('framenavigated', + lambda frame: asyncio.create_task(on_url_change()) if frame == page.main_frame else None) + try: + # 等待 URL 变化或超时 + await asyncio.wait_for(url_changed_event.wait(), timeout=200) # 最多等待 200 秒 + print("监听页面跳转成功") + except asyncio.TimeoutError: + print("监听页面跳转超时") + await page.close() + await context.close() + await browser.close() + status_queue.put("500") + return None + uuid_v1 = uuid.uuid1() + print(f"UUID v1: {uuid_v1}") + await context.storage_state(path=Path(BASE_DIR / "cookiesFile" / f"{uuid_v1}.json")) + result = await check_cookie(3, f"{uuid_v1}.json") + if not result: + status_queue.put("500") + await page.close() + await context.close() + await browser.close() + return None + await page.close() + await context.close() + await browser.close() + with sqlite3.connect(Path(BASE_DIR / "db" / "database.db")) as conn: + cursor = conn.cursor() + cursor.execute(''' + INSERT INTO user_info (type, filePath, userName, status) + VALUES (?, ?, ?, ?) + ''', (3, f"{uuid_v1}.json", id, 1)) + conn.commit() + print("✅ 用户状态已记录") + status_queue.put("200") + + +# 视频号登录 +async def get_tencent_cookie(id,status_queue): + url_changed_event = asyncio.Event() + async def on_url_change(): + # 检查是否是主框架的变化 + if page.url != original_url: + url_changed_event.set() + + async with async_playwright() as playwright: + options = { + 'args': [ + '--lang en-GB' + ], + 'headless': False, # Set headless option here + } + # Make sure to run headed. + browser = await playwright.chromium.launch(**options) + # Setup context however you like. + context = await browser.new_context() # Pass any options + # Pause the page, and start recording manually. + context = await set_init_script(context) + page = await context.new_page() + await page.goto("https://channels.weixin.qq.com") + original_url = page.url + + # 监听页面的 'framenavigated' 事件,只关注主框架的变化 + page.on('framenavigated', + lambda frame: asyncio.create_task(on_url_change()) if frame == page.main_frame else None) + + # 等待 iframe 出现(最多等 60 秒) + iframe_locator = page.frame_locator("iframe").first + + # 获取 iframe 中的第一个 img 元素 + img_locator = iframe_locator.get_by_role("img").first + + # 获取 src 属性值 + src = await img_locator.get_attribute("src") + print("✅ 图片地址:", src) + status_queue.put(src) + + try: + # 等待 URL 变化或超时 + await asyncio.wait_for(url_changed_event.wait(), timeout=200) # 最多等待 200 秒 + print("监听页面跳转成功") + except asyncio.TimeoutError: + status_queue.put("500") + print("监听页面跳转超时") + await page.close() + await context.close() + await browser.close() + return None + uuid_v1 = uuid.uuid1() + print(f"UUID v1: {uuid_v1}") + await context.storage_state(path=Path(BASE_DIR / "cookiesFile" / f"{uuid_v1}.json")) + result = await check_cookie(2,f"{uuid_v1}.json") + if not result: + status_queue.put("500") + await page.close() + await context.close() + await browser.close() + return None + await page.close() + await context.close() + await browser.close() + + with sqlite3.connect(Path(BASE_DIR / "db" / "database.db")) as conn: + cursor = conn.cursor() + cursor.execute(''' + INSERT INTO user_info (type, filePath, userName, status) + VALUES (?, ?, ?, ?) + ''', (2, f"{uuid_v1}.json", id, 1)) + conn.commit() + print("✅ 用户状态已记录") + status_queue.put("200") + +# 快手登录 +async def get_ks_cookie(id,status_queue): + url_changed_event = asyncio.Event() + async def on_url_change(): + # 检查是否是主框架的变化 + if page.url != original_url: + url_changed_event.set() + async with async_playwright() as playwright: + options = { + 'args': [ + '--lang en-GB' + ], + 'headless': False, # Set headless option here + } + # Make sure to run headed. + browser = await playwright.chromium.launch(**options) + # Setup context however you like. + context = await browser.new_context() # Pass any options + context = await set_init_script(context) + # Pause the page, and start recording manually. + page = await context.new_page() + await page.goto("https://cp.kuaishou.com") + + # 定位并点击“立即登录”按钮(类型为 link) + await page.get_by_role("link", name="立即登录").click() + await page.get_by_text("扫码登录").click() + img_locator = page.get_by_role("img", name="qrcode") + # 获取 src 属性值 + src = await img_locator.get_attribute("src") + original_url = page.url + print("✅ 图片地址:", src) + status_queue.put(src) + # 监听页面的 'framenavigated' 事件,只关注主框架的变化 + page.on('framenavigated', + lambda frame: asyncio.create_task(on_url_change()) if frame == page.main_frame else None) + + try: + # 等待 URL 变化或超时 + await asyncio.wait_for(url_changed_event.wait(), timeout=200) # 最多等待 200 秒 + print("监听页面跳转成功") + except asyncio.TimeoutError: + status_queue.put("500") + print("监听页面跳转超时") + await page.close() + await context.close() + await browser.close() + return None + uuid_v1 = uuid.uuid1() + print(f"UUID v1: {uuid_v1}") + await context.storage_state(path=Path(BASE_DIR / "cookiesFile" / f"{uuid_v1}.json")) + result = await check_cookie(4, f"{uuid_v1}.json") + if not result: + status_queue.put("500") + await page.close() + await context.close() + await browser.close() + return None + await page.close() + await context.close() + await browser.close() + + with sqlite3.connect(Path(BASE_DIR / "db" / "database.db")) as conn: + cursor = conn.cursor() + cursor.execute(''' + INSERT INTO user_info (type, filePath, userName, status) + VALUES (?, ?, ?, ?) + ''', (4, f"{uuid_v1}.json", id, 1)) + conn.commit() + print("✅ 用户状态已记录") + status_queue.put("200") +# +# a = asyncio.run(douyin_cookie_gen(2,None)) +# print(a) \ No newline at end of file