1. use loguru for clean console message

bilibili_uploader/main.py

@@ -3,6 +3,8 @@ import pathlib
 import random
 from biliup.plugins.bili_webup import BiliBili, Data
 
+from utils.log import bilibili_logger
+
 
 def extract_keys_from_json(data):
     """Extract specified keys from the provided JSON data."""
@@ -70,8 +72,8 @@ class BilibiliUploader(object):
             self.data.append(video_part)
             ret = bili.submit()  # 提交视频
             if ret.get('code') == 0:
-                print(f'[+] {self.file.name}上传 成功')
+                bilibili_logger.success(f'[+] {self.file.name}上传 成功')
                 return True
             else:
-                print(f'[-] {self.file.name}上传 失败, error messge: {ret.get("message")}')
+                bilibili_logger.error(f'[-] {self.file.name}上传 失败, error messge: {ret.get("message")}')
                 return False

douyin_uploader/main.py

@@ -7,6 +7,7 @@ import asyncio
 
 from conf import LOCAL_CHROME_PATH
 from utils.base_social_media import set_init_script
+from utils.log import douyin_logger
 
 
 async def cookie_auth(account_file):
@@ -20,10 +21,10 @@ async def cookie_auth(account_file):
         await page.goto("https://creator.douyin.com/creator-micro/content/upload")
         try:
             await page.wait_for_selector("div.boards-more h3:text('抖音排行榜')", timeout=5000)  # 等待5秒
-            print("[+] 等待5秒 cookie 失效")
+            douyin_logger.error(f'等待5秒 cookie 失效')
             return False
         except:
-            print("[+] cookie 有效")
+            douyin_logger.success(f'cookie 有效')
             return True
 
 
@@ -32,7 +33,7 @@ async def douyin_setup(account_file, handle=False):
         if not handle:
             # Todo alert message
             return False
-        print('[+] cookie文件不存在或已失效，即将自动打开浏览器，请扫码登录，登陆后会自动生成cookie文件')
+        douyin_logger.info('[+] cookie文件不存在或已失效，即将自动打开浏览器，请扫码登录，登陆后会自动生成cookie文件')
         await douyin_cookie_gen(account_file)
     return True
 
@@ -82,7 +83,7 @@ class DouYinVideo(object):
         await asyncio.sleep(1)
 
     async def handle_upload_error(self, page):
-        print("视频出错了，重新上传中")
+        douyin_logger.info('视频出错了，重新上传中')
         await page.locator('div.progress-div [class^="upload-btn-input"]').set_input_files(self.file_path)
 
     async def upload(self, playwright: Playwright) -> None:
@@ -99,9 +100,9 @@ class DouYinVideo(object):
         page = await context.new_page()
         # 访问指定的 URL
         await page.goto("https://creator.douyin.com/creator-micro/content/upload")
-        print('[+]正在上传-------{}.mp4'.format(self.title))
+        douyin_logger.info(f'[+]正在上传-------{self.title}.mp4')
         # 等待页面跳转到指定的 URL，没进入，则自动等待到超时
-        print('[-] 正在打开主页...')
+        douyin_logger.info(f'[-] 正在打开主页...')
         await page.wait_for_url("https://creator.douyin.com/creator-micro/content/upload")
         # 点击 "上传视频" 按钮
         await page.locator(".upload-btn--9eZLd").set_input_files(self.file_path)
@@ -114,32 +115,30 @@ class DouYinVideo(object):
                     "https://creator.douyin.com/creator-micro/content/publish?enter_from=publish_page")
                 break
             except:
-                print("  [-] 正在等待进入视频发布页面...")
+                douyin_logger.info(f'  [-] 正在等待进入视频发布页面...')
                 await asyncio.sleep(0.1)
 
         # 填充标题和话题
         # 检查是否存在包含输入框的元素
         # 这里为了避免页面变化，故使用相对位置定位：作品标题父级右侧第一个元素的input子元素
         await asyncio.sleep(1)
-        print("  [-] 正在填充标题和话题...")
+        douyin_logger.info(f'  [-] 正在填充标题和话题...')
         title_container = page.get_by_text('作品标题').locator("..").locator("xpath=following-sibling::div[1]").locator("input")
         if await title_container.count():
             await title_container.fill(self.title[:30])
         else:
             titlecontainer = page.locator(".notranslate")
             await titlecontainer.click()
-            print("clear existing title")
             await page.keyboard.press("Backspace")
             await page.keyboard.press("Control+KeyA")
             await page.keyboard.press("Delete")
-            print("filling new  title")
             await page.keyboard.type(self.title)
             await page.keyboard.press("Enter")
         css_selector = ".zone-container"
         for index, tag in enumerate(self.tags, start=1):
-            print("正在添加第%s个话题" % index)
             await page.type(css_selector, "#" + tag)
             await page.press(css_selector, "Space")
+        douyin_logger.info(f'总共添加{len(self.tags)}个话题')
 
         while True:
             # 判断重新上传按钮是否存在，如果不存在，代表视频正在上传，则等待
@@ -147,23 +146,22 @@ class DouYinVideo(object):
                 #  新版：定位重新上传
                 number = await page.locator('div label+div:has-text("重新上传")').count()
                 if number > 0:
-                    print("  [-]视频上传完毕")
+                    douyin_logger.success("  [-]视频上传完毕")
                     break
                 else:
-                    print("  [-] 正在上传视频中...")
+                    douyin_logger.info("  [-] 正在上传视频中...")
                     await asyncio.sleep(2)
 
                     if await page.locator('div.progress-div > div:has-text("上传失败")').count():
-                        print("  [-] 发现上传出错了...")
+                        douyin_logger.error("  [-] 发现上传出错了... 准备重试")
                         await self.handle_upload_error(page)
             except:
-                print("  [-] 正在上传视频中...")
+                douyin_logger.info("  [-] 正在上传视频中...")
                 await asyncio.sleep(2)
 
         # 更换可见元素
         await page.locator('div.semi-select span:has-text("输入地理位置")').click()
         await asyncio.sleep(1)
-        print("clear existing location")
         await page.keyboard.press("Backspace")
         await page.keyboard.press("Control+KeyA")
         await page.keyboard.press("Delete")
@@ -192,15 +190,15 @@ class DouYinVideo(object):
                     await publish_button.click()
                 await page.wait_for_url("https://creator.douyin.com/creator-micro/content/manage",
                                         timeout=1500)  # 如果自动跳转到作品页面，则代表发布成功
-                print("  [-]视频发布成功")
+                douyin_logger.success("  [-]视频发布成功")
                 break
             except:
-                print("  [-] 视频正在发布中...")
+                douyin_logger.info("  [-] 视频正在发布中...")
                 await page.screenshot(full_page=True)
                 await asyncio.sleep(0.5)
 
         await context.storage_state(path=self.account_file)  # 保存cookie
-        print('  [-]cookie更新完毕！')
+        douyin_logger.success('  [-]cookie更新完毕！')
         await asyncio.sleep(2)  # 这里延迟是为了方便眼睛直观的观看
         # 关闭浏览器上下文和浏览器实例
         await context.close()

requirements.txt

@@ -5,4 +5,5 @@ schedule
 cf_clearance
 biliup
 xhs
-qrcode
+qrcode
+loguru

tencent_uploader/main.py

@@ -8,6 +8,7 @@ import asyncio
 from conf import LOCAL_CHROME_PATH
 from utils.base_social_media import set_init_script
 from utils.files_times import get_absolute_path
+from utils.log import tencent_logger
 
 
 def format_str_for_short_title(origin_title: str) -> str:
@@ -41,10 +42,10 @@ async def cookie_auth(account_file):
         await page.goto("https://channels.weixin.qq.com/platform/post/create")
         try:
             await page.wait_for_selector('div.title-name:has-text("视频号小店")', timeout=5000)  # 等待5秒
-            print("[+] 等待5秒 cookie 失效")
+            tencent_logger.error("[+] 等待5秒 cookie 失效")
             return False
         except:
-            print("[+] cookie 有效")
+            tencent_logger.success("[+] cookie 有效")
             return True
 
 
@@ -75,7 +76,7 @@ async def weixin_setup(account_file, handle=False):
         if not handle:
             # Todo alert message
             return False
-        print('[+] cookie文件不存在或已失效，即将自动打开浏览器，请扫码登录，登陆后会自动生成cookie文件')
+        tencent_logger.info('[+] cookie文件不存在或已失效，即将自动打开浏览器，请扫码登录，登陆后会自动生成cookie文件')
         await get_tencent_cookie(account_file)
     return True
 
@@ -91,8 +92,6 @@ class TencentVideo(object):
         self.local_executable_path = LOCAL_CHROME_PATH
 
     async def set_schedule_time_tencent(self, page, publish_date):
-        print("click schedule")
-
         label_element = page.locator("label").filter(has_text="定时").nth(1)
         await label_element.click()
 
@@ -128,7 +127,7 @@ class TencentVideo(object):
         await page.locator("div.input-editor").click()
 
     async def handle_upload_error(self, page):
-        print("视频出错了，重新上传中")
+        tencent_logger.info("视频出错了，重新上传中")
         await page.locator('div.media-status-content div.tag-inner:has-text("删除")').click()
         await page.get_by_role('button', name="删除", exact=True).click()
         file_input = page.locator('input[type="file"]')
@@ -145,7 +144,7 @@ class TencentVideo(object):
         page = await context.new_page()
         # 访问指定的 URL
         await page.goto("https://channels.weixin.qq.com/platform/post/create")
-        print('[+]正在上传-------{}.mp4'.format(self.title))
+        tencent_logger.info(f'[+]正在上传-------{self.title}.mp4')
         # 等待页面跳转到指定的 URL，没进入，则自动等待到超时
         await page.wait_for_url("https://channels.weixin.qq.com/platform/post/create")
         # await page.wait_for_selector('input[type="file"]', timeout=10000)
@@ -160,7 +159,7 @@ class TencentVideo(object):
         # 原创选择
         await self.add_original(page)
         # 检测上传状态
-        await self.detact_upload_status(page)
+        await self.detect_upload_status(page)
         if self.publish_date != 0:
             await self.set_schedule_time_tencent(page, self.publish_date)
         # 添加短标题
@@ -169,12 +168,11 @@ class TencentVideo(object):
         await self.click_publish(page)
 
         await context.storage_state(path=f"{self.account_file}")  # 保存cookie
-        print('  [-]cookie更新完毕！')
+        tencent_logger.success('  [-]cookie更新完毕！')
         await asyncio.sleep(2)  # 这里延迟是为了方便眼睛直观的观看
         # 关闭浏览器上下文和浏览器实例
         await context.close()
         await browser.close()
-        print('[+]正在监控执行计划中.......')
 
     async def add_short_title(self, page):
         short_title_element = page.get_by_text("短标题", exact=True).locator("..").locator(
@@ -191,38 +189,37 @@ class TencentVideo(object):
                 if await publish_buttion.count():
                     await publish_buttion.click()
                 await page.wait_for_url("https://channels.weixin.qq.com/platform/post/list", timeout=1500)
-                print("  [-]视频发布成功")
+                tencent_logger.success("  [-]视频发布成功")
                 break
             except Exception as e:
                 current_url = page.url
                 if "https://channels.weixin.qq.com/platform/post/list" in current_url:
-                    print("  [-]视频发布成功")
+                    tencent_logger.success("  [-]视频发布成功")
                     break
                 else:
-                    print(f"  [-] Exception: {e}")
-                    print("  [-] 视频正在发布中...")
-                    await page.screenshot(full_page=True)
+                    tencent_logger.exception(f"  [-] Exception: {e}")
+                    tencent_logger.info("  [-] 视频正在发布中...")
                     await asyncio.sleep(0.5)
 
-    async def detact_upload_status(self, page):
+    async def detect_upload_status(self, page):
         while True:
             # 匹配删除按钮，代表视频上传完毕，如果不存在，��表视频正在上传，则等待
             try:
                 # 匹配删除按钮，代表视频上传完毕
                 if "weui-desktop-btn_disabled" not in await page.get_by_role("button", name="发表").get_attribute(
                         'class'):
-                    print("  [-]视频上传完毕")
+                    tencent_logger.info("  [-]视频上传完毕")
                     break
                 else:
-                    print("  [-] 正在上传视频中...")
+                    tencent_logger.info("  [-] 正在上传视频中...")
                     await asyncio.sleep(2)
                     # 出错了视频出错
                     if await page.locator('div.status-msg.error').count() and await page.locator(
                             'div.media-status-content div.tag-inner:has-text("删除")').count():
-                        print("  [-] 发现上传出错了...")
+                        tencent_logger.error("  [-] 发现上传出错了...准备重试")
                         await self.handle_upload_error(page)
             except:
-                print("  [-] 正在上传视频中...")
+                tencent_logger.info("  [-] 正在上传视频中...")
                 await asyncio.sleep(2)
 
     async def add_title_tags(self, page):
@@ -232,7 +229,7 @@ class TencentVideo(object):
         for index, tag in enumerate(self.tags, start=1):
             await page.keyboard.type("#" + tag)
             await page.keyboard.press("Space")
-        print(f"成功添加hashtag: {len(self.tags)}")
+        tencent_logger.info(f"成功添加hashtag: {len(self.tags)}")
 
     async def add_collection(self, page):
         collection_elements = page.get_by_text("添加到合集").locator("xpath=following-sibling::div").locator(

tk_uploader/main.py

@@ -8,6 +8,7 @@ import asyncio
 from 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 tiktok_logger
 
 
 async def cookie_auth(account_file):
@@ -27,12 +28,12 @@ async def cookie_auth(account_file):
                 class_name = await element.get_attribute('class')
                 # 使用正则表达式匹配特定模式的 class 名称
                 if re.match(r'tiktok-.*-SelectFormContainer.*', class_name):
-                    print("[+] cookie expired")
+                    tiktok_logger.error("[+] cookie expired")
                     return False
-            print("[+] cookie valid")
+            tiktok_logger.success("[+] cookie valid")
             return True
         except:
-            print("[+] cookie valid")
+            tiktok_logger.success("[+] cookie valid")
             return True
 
 
@@ -41,7 +42,7 @@ async def tiktok_setup(account_file, handle=False):
     if not os.path.exists(account_file) or not await cookie_auth(account_file):
         if not handle:
             return False
-        print('[+] 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')
+        tiktok_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_tiktok_cookie(account_file)
     return True
 
@@ -76,7 +77,6 @@ class TiktokVideo(object):
         self.account_file = account_file
 
     async def set_schedule_time(self, page, publish_date):
-        print("click schedule")
         schedule_input_element = page.frame_locator(Tk_Locator.tk_iframe).locator('div.scheduled-container input')
         await schedule_input_element.wait_for(state='visible')  # 确保按钮可见
 
@@ -130,7 +130,7 @@ class TiktokVideo(object):
         await page.frame_locator(Tk_Locator.tk_iframe).locator("h1:has-text('Upload video')").click()
 
     async def handle_upload_error(self, page):
-        print("video upload error retrying.")
+        tiktok_logger.info("video upload error retrying.")
         select_file_button = page.frame_locator(Tk_Locator.tk_iframe).locator('button[aria-label="Select file"]')
         async with page.expect_file_chooser() as fc_info:
             await select_file_button.click()
@@ -144,7 +144,7 @@ class TiktokVideo(object):
         page = await context.new_page()
 
         await page.goto("https://www.tiktok.com/creator-center/upload")
-        print('[+]Uploading-------{}.mp4'.format(self.title))
+        tiktok_logger.info(f'[+]Uploading-------{self.title}.mp4')
 
         await page.wait_for_url("https://www.tiktok.com/tiktokstudio/upload", timeout=10000)
         await page.wait_for_selector('iframe[data-tt="Upload_index_iframe"]')
@@ -166,7 +166,7 @@ class TiktokVideo(object):
         await self.click_publish(page)
 
         await context.storage_state(path=f"{self.account_file}")  # save cookie
-        print('  [-] update cookie！')
+        tiktok_logger.info('  [-] update cookie！')
         await asyncio.sleep(2)  # close delay for look the video status
         # close all
         await context.close()
@@ -195,7 +195,7 @@ class TiktokVideo(object):
 
         # tag part
         for index, tag in enumerate(self.tags, start=1):
-            print("Setting the %s tag" % index)
+            tiktok_logger.info("Setting the %s tag" % index)
             await page.keyboard.press("End")
             await page.wait_for_timeout(1000)  # 等待1秒
             await page.keyboard.insert_text("#" + tag + " ")
@@ -214,15 +214,15 @@ class TiktokVideo(object):
                     await publish_button.click()
 
                 await page.frame_locator(Tk_Locator.tk_iframe).locator(success_flag_div).wait_for(state="visible", timeout=1500)
-                print("  [-] video published success")
+                tiktok_logger.success("  [-] video published success")
                 break
             except Exception as e:
                 if await page.frame_locator(Tk_Locator.tk_iframe).locator(success_flag_div).count():
-                    print("  [-]video published success")
+                    tiktok_logger.success("  [-]video published success")
                     break
                 else:
-                    print(f"  [-] Exception: {e}")
-                    print("  [-] video publishing")
+                    tiktok_logger.exception(f"  [-] Exception: {e}")
+                    tiktok_logger.info("  [-] video publishing")
                     await page.screenshot(full_page=True)
                     await asyncio.sleep(0.5)
 
@@ -230,16 +230,16 @@ class TiktokVideo(object):
         while True:
             try:
                 if await page.frame_locator(Tk_Locator.tk_iframe).locator('div.btn-post > button').get_attribute("disabled") is None:
-                    print("  [-]video uploaded.")
+                    tiktok_logger.info("  [-]video uploaded.")
                     break
                 else:
-                    print("  [-] video uploading...")
+                    tiktok_logger.info("  [-] video uploading...")
                     await asyncio.sleep(2)
                     if await page.frame_locator(Tk_Locator.tk_iframe).locator('button[aria-label="Select file"]').count():
-                        print("  [-] found some error while uploading now retry...")
+                        tiktok_logger.info("  [-] found some error while uploading now retry...")
                         await self.handle_upload_error(page)
             except:
-                print("  [-] video uploading...")
+                tiktok_logger.info("  [-] video uploading...")
                 await asyncio.sleep(2)
 
     async def main(self):

utils/log.py

@@ -0,0 +1,50 @@
+from pathlib import Path
+from sys import stdout
+from loguru import logger
+
+from conf import BASE_DIR
+
+
+def log_formatter(record: dict) -> str:
+    """
+    Formatter for log records.
+    :param dict record: Log object containing log metadata & message.
+    :returns: str
+    """
+    colors = {
+        "TRACE": "#cfe2f3",
+        "INFO": "#9cbfdd",
+        "DEBUG": "#8598ea",
+        "WARNING": "#dcad5a",
+        "SUCCESS": "#3dd08d",
+        "ERROR": "#ae2c2c"
+    }
+    color = colors.get(record["level"].name, "#b3cfe7")
+    return f"<fg #70acde>{{time:YYYY-MM-DD HH:mm:ss}}</fg #70acde> | <fg {color}>{{level}}</fg {color}>: <light-white>{{message}}</light-white>\n"
+
+
+def create_logger(log_name: str, file_path: str):
+    """
+    Create custom logger for different business modules.
+    :param str log_name: name of log
+    :param str file_path: Optional path to log file
+    :returns: Configured logger
+    """
+    def filter_record(record):
+        return record["extra"].get("business_name") == log_name
+
+    Path(BASE_DIR / file_path).parent.mkdir(exist_ok=True)
+    logger.add(Path(BASE_DIR / file_path), filter=filter_record, level="INFO", rotation="10 MB", retention="10 days", backtrace=True, diagnose=True)
+    return logger.bind(business_name=log_name)
+
+
+# Remove all existing handlers
+logger.remove()
+# Add a standard console handler
+logger.add(stdout, colorize=True, format=log_formatter)
+
+douyin_logger = create_logger('douyin', 'logs/douyin.log')
+tencent_logger = create_logger('tencent', 'logs/tencent.log')
+xhs_logger = create_logger('xhs', 'logs/xhs.log')
+tiktok_logger = create_logger('tiktok', 'logs/tiktok.log')
+bilibili_logger = create_logger('bilibili', 'logs/bilibili.log')