from pathlib import Path import sys import os # 添加项目根目录到 Python 路径 current_dir = Path(__file__).parent.resolve() project_root = current_dir.parent.parent sys.path.append(str(project_root)) from conf import LOCAL_CHROME_PATH, BASE_DIR import re from datetime import datetime from playwright.async_api import Playwright, async_playwright import os import asyncio #from uploader.tk_uploader.tk_config import Tk_Locator from utils.base_social_media import set_init_script from utils.files_times import get_absolute_path from utils.log import X_logger from utils.anti_detection import create_stealth_browser, create_stealth_context, setup_stealth_page async def cookie_auth(account_file): async with async_playwright() as playwright: browser = await playwright.chromium.launch(headless=True) context = await browser.new_context(storage_state=account_file) context = await set_init_script(context) # 创建一个新的页面 page = await context.new_page() # 访问指定的 URL await page.goto("https://x.com/i/flow/login") #await page.wait_for_load_state('networkidle') await page.wait_for_load_state('domcontentloaded', timeout=30000) try: # 选择所有的 select 元素 select_elements = await page.query_selector_all('select') for element in select_elements: class_name = await element.get_attribute('class') # 使用正则表达式匹配特定模式的 class 名称 if re.match(r'ins-.*-SelectFormContainer.*', class_name): X_logger.error("[+] cookie expired") return False X_logger.success("[+] cookie valid") return True except: X_logger.success("[+] cookie valid") return True async def X_setup(account_file, handle=False): account_file = get_absolute_path(account_file, "X_uploader") if not os.path.exists(account_file) or not await cookie_auth(account_file): if not handle: return False X_logger.info('[+] cookie file is not existed or expired. Now open the browser auto. Please login with your way(gmail phone, whatever, the cookie file will generated after login') await get_X_cookie(account_file) return True async def get_X_cookie(account_file): 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://x.com/") await page.pause() # 点击调试器的继续，保存cookie await context.storage_state(path=account_file) class XVideo(object): def __init__(self, title, file_path, tags, publish_date, account_file, thumbnail_path=None, headless=True): self.title = title self.file_path = file_path self.tags = tags self.publish_date = publish_date self.thumbnail_path = thumbnail_path self.account_file = account_file self.local_executable_path = LOCAL_CHROME_PATH self.headless = headless # 是否使用无头模式，默认为True self.locator_base = None async def handle_upload_error(self, page): X_logger.info("X(Twitter)上传错误，正在重试...") try: # 重新查找文件上传元素并重试 upload_element = page.locator('input[type="file"]') await upload_element.wait_for(state='visible', timeout=10000) await upload_element.set_input_files(self.file_path) X_logger.info("重试上传成功") except Exception as e: X_logger.error(f"重试上传失败: {e}") async def upload(self, playwright: Playwright) -> None: # 使用统一的反检测工具 browser = await create_stealth_browser( playwright=playwright, headless=self.headless, executable_path=self.local_executable_path ) context = await create_stealth_context( browser=browser, account_file=self.account_file, headless=self.headless ) page = await setup_stealth_page(context, "https://x.com/") X_logger.info(f'[+]Uploading to X(Twitter)-------{self.title}') await page.wait_for_load_state('domcontentloaded', timeout=30000) try: # 基础的X上传逻辑（需要根据实际页面结构完善） X_logger.info("X上传器基础框架已就绪，需要根据实际页面结构完善上传逻辑") # TODO: 实现具体的上传逻辑 # 1. 点击发推按钮 # 2. 上传媒体文件 # 3. 添加文本内容 # 4. 发布推文 except Exception as e: X_logger.error(f"上传过程中出现错误: {e}") await self.handle_upload_error(page) await context.storage_state(path=f"{self.account_file}") # save cookie X_logger.info(' [-] 更新cookie成功！') await asyncio.sleep(2) # close all await context.close() await browser.close() async def main(self): async with async_playwright() as playwright: await self.upload(playwright) async def main(): """用于测试X上传功能的主函数""" from pathlib import Path # 配置文件路径 account_file = Path(__file__).parent.parent.parent / "cookies" / "X_uploader" / "account.json" account_file.parent.mkdir(exist_ok=True, parents=True) # 首先确保cookie有效 if not await X_setup(str(account_file), handle=True): X_logger.error("Cookie设置失败") return # 创建测试视频对象 video_path = Path(__file__).parent.parent.parent / "videos" / "demo.mp4" if not video_path.exists(): X_logger.error(f"测试视频文件不存在: {video_path}") return x_video = XVideo( title="测试X(Twitter)上传", file_path=str(video_path), tags=["test", "twitter", "upload"], publish_date=datetime.now(), account_file=str(account_file), headless=False # 推荐使用有头模式 ) # 开始上传 await x_video.main() if __name__ == '__main__': account_file = Path(BASE_DIR / "cookies" / "X_uploader" / "account.json") account_file.parent.mkdir(exist_ok=True) cookie_setup = asyncio.run(X_setup(str(account_file), handle=True)) # 如果需要测试上传功能，取消下面的注释 # asyncio.run(main()) # 以下是旧的注释代码，保留作为参考 # class InsVideo(object): # def __init__(self, title, file_path, tags, publish_date, account_file, thumbnail_path=None): # self.title = title # self.file_path = file_path # self.tags = tags # self.publish_date = publish_date # self.thumbnail_path = thumbnail_path # self.account_file = account_file # self.local_executable_path = LOCAL_CHROME_PATH # self.locator_base = None # # click title to remove the focus. # # await self.locator_base.locator("h1:has-text('Upload video')").click() # async def handle_upload_error(self, page): # X_logger.info("Instagram上传错误，正在重试...") # try: # # 重新查找"从电脑中选择"按钮并重试 # select_from_computer_button = page.locator('button._aswp._aswr._aswu._asw_._asx2:has-text("从电脑中选择")') # await select_from_computer_button.wait_for(state='visible', timeout=10000) # async with page.expect_file_chooser() as fc_info: # await select_from_computer_button.click() # file_chooser = await fc_info.value # await file_chooser.set_files(self.file_path) # X_logger.info("重试上传成功") # except Exception as e: # X_logger.error(f"重试上传失败: {e}") # async def upload(self, playwright: Playwright) -> None: # browser = await playwright.chromium.launch(headless=False, executable_path=self.local_executable_path) # context = await browser.new_context(storage_state=f"{self.account_file}") # context = await set_init_script(context) # page = await context.new_page() # # 访问Instagram主页 # await page.goto("https://www.instagram.com/") # X_logger.info(f'[+]Uploading to Instagram-------{self.title}') # await page.wait_for_load_state('domcontentloaded', timeout=30000) # try: # # 点击创建按钮 # create_button = page.locator('span.x1lliihq.x193iq5w.x6ikm8r.x10wlt62.xlyipyv.xuxw1ft:has-text("创建")') # await create_button.wait_for(state='visible', timeout=10000) # await create_button.click() # X_logger.info("成功点击创建按钮") # # 等待弹窗出现并点击"从电脑中选择" # await page.wait_for_timeout(2000) # 等待弹窗加载 # select_from_computer_button = page.locator('button._aswp._aswr._aswu._asw_._asx2:has-text("从电脑中选择")') # await select_from_computer_button.wait_for(state='visible', timeout=10000) # # 选择文件 # async with page.expect_file_chooser() as fc_info: # await select_from_computer_button.click() # file_chooser = await fc_info.value # await file_chooser.set_files(self.file_path) # X_logger.info("成功选择文件") # # 等待文件上传完成 # await page.wait_for_timeout(5000) # select_crop_button = page.locator('role=button >> text=继续') # await select_crop_button.wait_for(state='visible', timeout=10000) # await select_crop_button.click() # select_crop_button = page.locator('role=button >> text=继续') # await select_crop_button.wait_for(state='visible', timeout=10000) # await select_crop_button.click() # # 添加标题和标签 # await self.add_title_tags(page) # # 发布内容 # await self.click_publish(page) # except Exception as e: # X_logger.error(f"上传过程中出现错误: {e}") # # 处理上传错误 # await self.handle_upload_error(page) # await context.storage_state(path=f"{self.account_file}") # save cookie # X_logger.info(' [-] 更新cookie成功！') # await asyncio.sleep(2) # close delay for look the video status # # close all # await context.close() # await browser.close() # async def add_title_tags(self, page): # try: # # 等待Instagram的文本输入框出现 # await page.wait_for_timeout(3000) # 等待页面加载 # # 查找Instagram的文本输入框（可能有多种选择器） # text_selectors = [ # 'textarea[aria-label="写下说明…"]', # 'textarea[placeholder="写下说明…"]', # 'div[contenteditable="true"]', # 'textarea' # ] # text_area = None # for selector in text_selectors: # try: # text_area = page.locator(selector).first # await text_area.wait_for(state='visible', timeout=5000) # break # except: # continue # if text_area: # await text_area.click() # await page.wait_for_timeout(1000) # # 清空现有内容 # await page.keyboard.press("Control+A") # await page.keyboard.press("Delete") # # 输入标题 # await page.keyboard.insert_text(self.title) # await page.keyboard.press("Enter") # await page.keyboard.press("Enter") # # 添加标签 # for index, tag in enumerate(self.tags, start=1): # X_logger.info(f"添加第 {index} 个标签: {tag}") # await page.keyboard.insert_text(f"#{tag} ") # await page.wait_for_timeout(500) # X_logger.info("成功添加标题和标签") # else: # X_logger.warning("未找到文本输入框，跳过添加标题和标签") # # 调试用：暂停10秒，让你有时间观察页面上的标签是否正确输入 # await page.wait_for_timeout(10000) # X_logger.info("观察时间结束，中断程序") # # 强制中断（后续代码不执行） # raise SystemExit("调试中断：标签输入流程完成") # except Exception as e: # X_logger.error(f"添加标题和标签时出错: {e}") # async def click_publish(self, page): # try: # # 等待分享/发布按钮出现 # await page.wait_for_timeout(2000) # # Instagram发布按钮的可能选择器 # publish_selectors = [ # 'button:has-text("分享")', # 'button:has-text("Share")', # 'button:has-text("发布")', # 'button:has-text("Post")', # 'button[type="submit"]' # ] # publish_button = None # for selector in publish_selectors: # try: # publish_button = page.locator(selector).first # await publish_button.wait_for(state='visible', timeout=5000) # break # except: # continue # if publish_button: # await publish_button.click() # X_logger.info("成功点击发布按钮") # # 等待发布完成 # await page.wait_for_timeout(5000) # X_logger.success("Instagram内容发布成功！") # else: # X_logger.warning("未找到发布按钮") # except Exception as e: # X_logger.error(f"发布时出错: {e}") # async def main(): # """用于测试Instagram上传功能的主函数""" # from pathlib import Path # # 配置文件路径 # account_file = Path(__file__).parent.parent.parent / "cookies" / "ins_uploader" / "account.json" # account_file.parent.mkdir(exist_ok=True, parents=True) # # 首先确保cookie有效 # if not await ins_setup(str(account_file), handle=True): # X_logger.error("Cookie设置失败") # return # # 创建测试视频对象 # video_path = Path(__file__).parent.parent.parent / "videos" / "demo.mp4" # if not video_path.exists(): # X_logger.error(f"测试视频文件不存在: {video_path}") # return # ins_video = InsVideo( # title="测试Instagram上传", # file_path=str(video_path), # tags=["test", "instagram", "upload"], # publish_date=datetime.now(), # account_file=str(account_file) # ) # # 开始上传 # async with async_playwright() as playwright: # await ins_video.upload(playwright) # if __name__ == '__main__': # asyncio.run(main())