from abc import ABC, abstractmethod
|
from playwright.async_api import async_playwright
|
from auto_media_publisher.config.conf_base import COOKIE_DIR,HIDE_UPLOAD_PAGE,DEFAULT_USER_AGENT,PLAYWRIGHT_STEALTH_JS_PATH
|
from auto_media_publisher.utils.logger import get_logger
|
from pathlib import Path
|
import os
|
from datetime import datetime
|
|
class UploaderBase(ABC):
|
def __init__(self, user, platform, proxy):
|
self.user = user
|
self.platform = platform
|
self.proxy = proxy
|
self.cookie_path = os.path.join(COOKIE_DIR, self.user, f"{self.platform}.json")
|
self.logger = get_logger(self.platform, self.user)
|
self.playwright = None
|
|
@classmethod
|
def create_sub_class(cls, user,platform,proxy):
|
return cls(user, platform, proxy)
|
|
|
@abstractmethod
|
def _upload_core(self):
|
"""
|
上传视频的核心方法，需要子类实现。
|
"""
|
pass
|
|
async def _init_broswer_context(self):
|
self.playwright = await async_playwright().start()
|
browser_args = {"headless": HIDE_UPLOAD_PAGE,"channel": "chrome"}
|
self.logger.info(f"tring to init browser and upload, Using Proxy : {self.proxy}")
|
|
if isinstance(self.proxy, str) and self.proxy.strip().lower().startswith(('http://', 'https://')):
|
browser_args["proxy"] = {"server": self.proxy}
|
self.browser = await self.playwright.chromium.launch(**browser_args)
|
if not os.path.exists(self.cookie_path):
|
self.logger.info(f"{ self.cookie_path} DOES NOT exist. Please Try login first.")
|
return
|
self.context = await self.browser.new_context(storage_state=f"{self.cookie_path}", user_agent=DEFAULT_USER_AGENT,viewport={"width": 1920, "height": 1080})
|
if os.path.exists(PLAYWRIGHT_STEALTH_JS_PATH):
|
await self.context.add_init_script(path=Path(PLAYWRIGHT_STEALTH_JS_PATH))
|
await self.context.grant_permissions(['geolocation'])
|
|
|
async def colse_context_browser(self):
|
try:
|
if self.context:
|
await self.context.close()
|
self.context = None # 可选：释放引用
|
if self.browser:
|
await self.browser.close()
|
self.browser = None
|
if self.playwright:
|
await self.playwright.stop()
|
self.playwright = None
|
except Exception as e:
|
self.logger.error(f"关闭浏览器资源时发生错误: {e}")
|
|
|
async def upload(self, task_info: dict):
|
if not self.playwright:
|
await self._init_broswer_context()
|
for key, value in task_info.items():
|
setattr(self, key, value)
|
# 尝试转换 scheduled_time 为 datetime 对象
|
if hasattr(self, 'scheduled_time') and isinstance(self.scheduled_time, str):
|
try:
|
self.scheduled_time = datetime.strptime(self.scheduled_time, "%Y-%m-%d %H:%M:%S")
|
except Exception as e:
|
self.logger.error(f"{self.user},{self.platform}发生错误,日期转换 {self.scheduled_time},{e}")
|
try:
|
return await self._upload_core()
|
except Exception as e:
|
self.logger.error(f"Upload failed: {e}", exc_info=True)
|
return False
|
|
async def upload_and_close(self,task_info:dict):
|
result = await self.upload(task_info)
|
await self.colse_context_browser()
|
return result
|
|
