| 1 | import asyncio |
| 2 | import sqlite3 |
| 3 | |
| 4 | from playwright.async_api import async_playwright |
| 5 | |
| 6 | from myUtils.auth import check_cookie |
| 7 | from utils.base_social_media import set_init_script |
| 8 | import uuid |
| 9 | from pathlib import Path |
| 10 | from conf import BASE_DIR, LOCAL_CHROME_HEADLESS, LOCAL_CHROME_PATH |
| 11 | |
| 12 | # 统一获取浏览器启动配置（防风控+引入本地浏览器） |
| 13 | def get_browser_options(): |
| 14 | options = { |
| 15 | 'headless': LOCAL_CHROME_HEADLESS, |
| 16 | 'args': [ |
| 17 | '--disable-blink-features=AutomationControlled', # 核心防爬屏蔽：去掉 window.navigator.webdriver 标签 |
| 18 | '--lang=zh-CN', |
| 19 | '--disable-infobars', |
| 20 | '--start-maximized' |
| 21 | ] |
| 22 | } |
| 23 | # 如果用户在 conf.py 里配置了本地 Chrome，就用本地的，这样成功率极高 |
| 24 | if LOCAL_CHROME_PATH: |
| 25 | options['executable_path'] = LOCAL_CHROME_PATH |
| 26 | |
| 27 | return options |
| 28 | |
| 29 | # 抖音登录 |
| 30 | async def douyin_cookie_gen(id,status_queue): |
| 31 | url_changed_event = asyncio.Event() |
| 32 | async def on_url_change(): |
| 33 | # 检查是否是主框架的变化 |
| 34 | if page.url != original_url: |
| 35 | url_changed_event.set() |
| 36 | async with async_playwright() as playwright: |
| 37 | options = get_browser_options() |
| 38 | # Make sure to run headed. |
| 39 | browser = await playwright.chromium.launch(**options) |
| 40 | # Setup context however you like. |
| 41 | context = await browser.new_context() # Pass any options |
| 42 | context = await set_init_script(context) |
| 43 | # Pause the page, and start recording manually. |
| 44 | page = await context.new_page() |
| 45 | await page.goto("https://creator.douyin.com/") |
| 46 | original_url = page.url |
| 47 | img_locator = page.get_by_role("img", name="二维码") |
| 48 | # 获取 src 属性值 |
| 49 | src = await img_locator.get_attribute("src") |
| 50 | print("✅ 图片地址:", src) |
| 51 | status_queue.put(src) |
| 52 | # 监听页面的 'framenavigated' 事件，只关注主框架的变化 |
| 53 | page.on('framenavigated', |
| 54 | lambda frame: asyncio.create_task(on_url_change()) if frame == page.main_frame else None) |
| 55 | try: |
| 56 | # 等待 URL 变化或超时 |
| 57 | await asyncio.wait_for(url_changed_event.wait(), timeout=200) # 最多等待 200 秒 |
| 58 | print("监听页面跳转成功") |
| 59 | except asyncio.TimeoutError: |
| 60 | print("监听页面跳转超时") |
| 61 | await page.close() |
| 62 | await context.close() |
| 63 | await browser.close() |
| 64 | status_queue.put("500") |
| 65 | return None |
| 66 | uuid_v1 = uuid.uuid1() |
| 67 | print(f"UUID v1: {uuid_v1}") |
| 68 | # 确保cookiesFile目录存在 |
| 69 | cookies_dir = Path(BASE_DIR / "cookiesFile") |
| 70 | cookies_dir.mkdir(exist_ok=True) |
| 71 | await context.storage_state(path=cookies_dir / f"{uuid_v1}.json") |
| 72 | result = await check_cookie(3, f"{uuid_v1}.json") |
| 73 | if not result: |
| 74 | status_queue.put("500") |
| 75 | await page.close() |
| 76 | await context.close() |
| 77 | await browser.close() |
| 78 | return None |
| 79 | await page.close() |
| 80 | await context.close() |
| 81 | await browser.close() |
| 82 | with sqlite3.connect(Path(BASE_DIR / "db" / "database.db")) as conn: |
| 83 | cursor = conn.cursor() |
| 84 | cursor.execute(''' |
| 85 | INSERT INTO user_info (type, filePath, userName, status) |
| 86 | VALUES (?, ?, ?, ?) |
| 87 | ''', (3, f"{uuid_v1}.json", id, 1)) |
| 88 | conn.commit() |
| 89 | print("✅ 用户状态已记录") |
| 90 | status_queue.put("200") |
| 91 | |
| 92 | |
| 93 | # 视频号登录 |
| 94 | async def get_tencent_cookie(id,status_queue): |
| 95 | url_changed_event = asyncio.Event() |
| 96 | async def on_url_change(): |
| 97 | # 检查是否是主框架的变化 |
| 98 | if page.url != original_url: |
| 99 | url_changed_event.set() |
| 100 | |
| 101 | async with async_playwright() as playwright: |
| 102 | options = { |
| 103 | 'args': [ |
| 104 | '--lang en-GB' |
| 105 | ], |
| 106 | 'headless': LOCAL_CHROME_HEADLESS, # Set headless option here |
| 107 | } |
| 108 | # Make sure to run headed. |
| 109 | browser = await playwright.chromium.launch(**options) |
| 110 | # Setup context however you like. |
| 111 | context = await browser.new_context() # Pass any options |
| 112 | # Pause the page, and start recording manually. |
| 113 | context = await set_init_script(context) |
| 114 | page = await context.new_page() |
| 115 | await page.goto("https://channels.weixin.qq.com") |
| 116 | original_url = page.url |
| 117 | |
| 118 | # 监听页面的 'framenavigated' 事件，只关注主框架的变化 |
| 119 | page.on('framenavigated', |
| 120 | lambda frame: asyncio.create_task(on_url_change()) if frame == page.main_frame else None) |
| 121 | |
| 122 | # 等待 iframe 出现（最多等 60 秒） |
| 123 | iframe_locator = page.frame_locator("iframe").first |
| 124 | |
| 125 | # 获取 iframe 中的第一个 img 元素 |
| 126 | img_locator = iframe_locator.get_by_role("img").first |
| 127 | |
| 128 | # 获取 src 属性值 |
| 129 | src = await img_locator.get_attribute("src") |
| 130 | print("✅ 图片地址:", src) |
| 131 | status_queue.put(src) |
| 132 | |
| 133 | try: |
| 134 | # 等待 URL 变化或超时 |
| 135 | await asyncio.wait_for(url_changed_event.wait(), timeout=200) # 最多等待 200 秒 |
| 136 | print("监听页面跳转成功") |
| 137 | except asyncio.TimeoutError: |
| 138 | status_queue.put("500") |
| 139 | print("监听页面跳转超时") |
| 140 | await page.close() |
| 141 | await context.close() |
| 142 | await browser.close() |
| 143 | return None |
| 144 | uuid_v1 = uuid.uuid1() |
| 145 | print(f"UUID v1: {uuid_v1}") |
| 146 | # 确保cookiesFile目录存在 |
| 147 | cookies_dir = Path(BASE_DIR / "cookiesFile") |
| 148 | cookies_dir.mkdir(exist_ok=True) |
| 149 | await context.storage_state(path=cookies_dir / f"{uuid_v1}.json") |
| 150 | result = await check_cookie(2,f"{uuid_v1}.json") |
| 151 | if not result: |
| 152 | status_queue.put("500") |
| 153 | await page.close() |
| 154 | await context.close() |
| 155 | await browser.close() |
| 156 | return None |
| 157 | await page.close() |
| 158 | await context.close() |
| 159 | await browser.close() |
| 160 | |
| 161 | with sqlite3.connect(Path(BASE_DIR / "db" / "database.db")) as conn: |
| 162 | cursor = conn.cursor() |
| 163 | cursor.execute(''' |
| 164 | INSERT INTO user_info (type, filePath, userName, status) |
| 165 | VALUES (?, ?, ?, ?) |
| 166 | ''', (2, f"{uuid_v1}.json", id, 1)) |
| 167 | conn.commit() |
| 168 | print("✅ 用户状态已记录") |
| 169 | status_queue.put("200") |
| 170 | |
| 171 | # 快手登录 |
| 172 | async def get_ks_cookie(id,status_queue): |
| 173 | url_changed_event = asyncio.Event() |
| 174 | async def on_url_change(): |
| 175 | # 检查是否是主框架的变化 |
| 176 | if page.url != original_url: |
| 177 | url_changed_event.set() |
| 178 | async with async_playwright() as playwright: |
| 179 | options = { |
| 180 | 'args': [ |
| 181 | '--lang en-GB' |
| 182 | ], |
| 183 | 'headless': LOCAL_CHROME_HEADLESS, # Set headless option here |
| 184 | } |
| 185 | # Make sure to run headed. |
| 186 | browser = await playwright.chromium.launch(**options) |
| 187 | # Setup context however you like. |
| 188 | context = await browser.new_context() # Pass any options |
| 189 | context = await set_init_script(context) |
| 190 | # Pause the page, and start recording manually. |
| 191 | page = await context.new_page() |
| 192 | await page.goto("https://cp.kuaishou.com") |
| 193 | |
| 194 | # 定位并点击“立即登录”按钮（类型为 link） |
| 195 | await page.get_by_role("link", name="立即登录").click() |
| 196 | await page.get_by_text("扫码登录").click() |
| 197 | img_locator = page.get_by_role("img", name="qrcode") |
| 198 | # 获取 src 属性值 |
| 199 | src = await img_locator.get_attribute("src") |
| 200 | original_url = page.url |
| 201 | print("✅ 图片地址:", src) |
| 202 | status_queue.put(src) |
| 203 | # 监听页面的 'framenavigated' 事件，只关注主框架的变化 |
| 204 | page.on('framenavigated', |
| 205 | lambda frame: asyncio.create_task(on_url_change()) if frame == page.main_frame else None) |
| 206 | |
| 207 | try: |
| 208 | # 等待 URL 变化或超时 |
| 209 | await asyncio.wait_for(url_changed_event.wait(), timeout=200) # 最多等待 200 秒 |
| 210 | print("监听页面跳转成功") |
| 211 | except asyncio.TimeoutError: |
| 212 | status_queue.put("500") |
| 213 | print("监听页面跳转超时") |
| 214 | await page.close() |
| 215 | await context.close() |
| 216 | await browser.close() |
| 217 | return None |
| 218 | uuid_v1 = uuid.uuid1() |
| 219 | print(f"UUID v1: {uuid_v1}") |
| 220 | # 确保cookiesFile目录存在 |
| 221 | cookies_dir = Path(BASE_DIR / "cookiesFile") |
| 222 | cookies_dir.mkdir(exist_ok=True) |
| 223 | await context.storage_state(path=cookies_dir / f"{uuid_v1}.json") |
| 224 | result = await check_cookie(4, f"{uuid_v1}.json") |
| 225 | if not result: |
| 226 | status_queue.put("500") |
| 227 | await page.close() |
| 228 | await context.close() |
| 229 | await browser.close() |
| 230 | return None |
| 231 | await page.close() |
| 232 | await context.close() |
| 233 | await browser.close() |
| 234 | |
| 235 | with sqlite3.connect(Path(BASE_DIR / "db" / "database.db")) as conn: |
| 236 | cursor = conn.cursor() |
| 237 | cursor.execute(''' |
| 238 | INSERT INTO user_info (type, filePath, userName, status) |
| 239 | VALUES (?, ?, ?, ?) |
| 240 | ''', (4, f"{uuid_v1}.json", id, 1)) |
| 241 | conn.commit() |
| 242 | print("✅ 用户状态已记录") |
| 243 | status_queue.put("200") |
| 244 | |
| 245 | # 小红书登录 |
| 246 | async def xiaohongshu_cookie_gen(id,status_queue): |
| 247 | url_changed_event = asyncio.Event() |
| 248 | |
| 249 | async def on_url_change(): |
| 250 | # 检查是否是主框架的变化 |
| 251 | if page.url != original_url: |
| 252 | url_changed_event.set() |
| 253 | |
| 254 | async with async_playwright() as playwright: |
| 255 | options = { |
| 256 | 'args': [ |
| 257 | '--lang en-GB' |
| 258 | ], |
| 259 | 'headless': LOCAL_CHROME_HEADLESS, # Set headless option here |
| 260 | } |
| 261 | # Make sure to run headed. |
| 262 | browser = await playwright.chromium.launch(**options) |
| 263 | # Setup context however you like. |
| 264 | context = await browser.new_context() # Pass any options |
| 265 | context = await set_init_script(context) |
| 266 | # Pause the page, and start recording manually. |
| 267 | page = await context.new_page() |
| 268 | await page.goto("https://creator.xiaohongshu.com/") |
| 269 | await page.locator('img.css-wemwzq').click() |
| 270 | |
| 271 | img_locator = page.get_by_role("img").nth(2) |
| 272 | # 获取 src 属性值 |
| 273 | src = await img_locator.get_attribute("src") |
| 274 | original_url = page.url |
| 275 | print("✅ 图片地址:", src) |
| 276 | status_queue.put(src) |
| 277 | # 监听页面的 'framenavigated' 事件，只关注主框架的变化 |
| 278 | page.on('framenavigated', |
| 279 | lambda frame: asyncio.create_task(on_url_change()) if frame == page.main_frame else None) |
| 280 | |
| 281 | try: |
| 282 | # 等待 URL 变化或超时 |
| 283 | await asyncio.wait_for(url_changed_event.wait(), timeout=200) # 最多等待 200 秒 |
| 284 | print("监听页面跳转成功") |
| 285 | except asyncio.TimeoutError: |
| 286 | status_queue.put("500") |
| 287 | print("监听页面跳转超时") |
| 288 | await page.close() |
| 289 | await context.close() |
| 290 | await browser.close() |
| 291 | return None |
| 292 | uuid_v1 = uuid.uuid1() |
| 293 | print(f"UUID v1: {uuid_v1}") |
| 294 | # 确保cookiesFile目录存在 |
| 295 | cookies_dir = Path(BASE_DIR / "cookiesFile") |
| 296 | cookies_dir.mkdir(exist_ok=True) |
| 297 | await context.storage_state(path=cookies_dir / f"{uuid_v1}.json") |
| 298 | result = await check_cookie(1, f"{uuid_v1}.json") |
| 299 | if not result: |
| 300 | status_queue.put("500") |
| 301 | await page.close() |
| 302 | await context.close() |
| 303 | await browser.close() |
| 304 | return None |
| 305 | await page.close() |
| 306 | await context.close() |
| 307 | await browser.close() |
| 308 | |
| 309 | with sqlite3.connect(Path(BASE_DIR / "db" / "database.db")) as conn: |
| 310 | cursor = conn.cursor() |
| 311 | cursor.execute(''' |
| 312 | INSERT INTO user_info (type, filePath, userName, status) |
| 313 | VALUES (?, ?, ?, ?) |
| 314 | ''', (1, f"{uuid_v1}.json", id, 1)) |
| 315 | conn.commit() |
| 316 | print("✅ 用户状态已记录") |
| 317 | status_queue.put("200") |
| 318 | |
| 319 | # a = asyncio.run(xiaohongshu_cookie_gen(4,None)) |
| 320 | # print(a) |
| 321 |