fix bilibili is_login

_playwright/[no_use]aliyundrive.py

@@ -1,79 +0,0 @@
-# coding:utf-8
-import sys
-import asyncio
-
-from os.path import abspath, dirname, join
-from playwright.async_api import async_playwright
-
-sys.path.append(join(dirname(abspath(__file__)), "../"))
-
-from utils import push_msg
-from __init__ import (
-    browser_headless,
-    browser_proxy,
-    aliyundrive_url,
-    aliyundrive_cookie_file,
-    aliyundrive_title,
-    aliyundrive_logger,
-)
-
-
-# 判断登录状态
-async def is_login(page):
-    await page.goto(aliyundrive_url)
-    # page.locator("li").filter(has_text="回收站").get_by_role("img").click()
-    if page.get_by_text("回收站").is_visible():  # 加 await 就会在页面加载完成前跳过
-        aliyundrive_logger.info("登录成功")
-        return True
-    else:
-        stderr = "Cookie 过期"
-        aliyundrive_logger.error(stderr)
-        push_msg(aliyundrive_title, stderr)
-    return False
-
-
-# 签到
-async def sign(page):
-    msg = []
-
-    if not await is_login(page):
-        return "aliyundrive_not_login"
-
-    # await page.pause()
-
-    if page.get_by_text("待领取").is_visible():
-        sign_res = "今日已签到"
-        aliyundrive_logger.info(sign_res)
-        msg.append(sign_res)
-        # 领取奖励
-        await page.get_by_text("待领取").click()
-        reward_res = "今日领取奖励成功"
-        aliyundrive_logger.info(reward_res)
-    elif page.get_by_text("待签到").is_visible():
-        await page.get_by_text("待签到").click()
-        sign_res = "签到成功"
-        aliyundrive_logger.info(sign_res)
-        msg.append(sign_res)
-    else:
-        reward_res = "今日奖励已领取"
-        aliyundrive_logger.info(reward_res)
-    return "\n".join(msg)
-
-
-async def main():
-    async with async_playwright() as playwright:
-        browser = await playwright.chromium.launch(headless=browser_headless)
-        context = await browser.new_context(storage_state=aliyundrive_cookie_file)
-        page = await context.new_page()
-        task = asyncio.create_task(sign(page))
-        await asyncio.gather(task)
-        # 退出
-        await page.close()
-        await browser.close()
-        # 推送
-        result = task.result()
-        push_msg(aliyundrive_title, "\n".join([result]))
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

_playwright/bilibili.py

@@ -7,7 +7,7 @@ from os.path import abspath, dirname, join
 from playwright.async_api import async_playwright
 
 sys.path.append(join(dirname(abspath(__file__)), "../"))
-from utils import push_msg, get_bilibili_live_rooms_from_pi
+from utils import push_msg
 from __init__ import (
     browser_headless,
     browser_proxy,
@@ -19,102 +19,112 @@ from __init__ import (
     bilibili_logger,
     bilibili_img,
 )
+from option_mysql import get_aiomysql_instance
+
+
+async def get_bilibili_live_rooms_from_pi() -> dict:
+    sql = f"SELECT room_id, username FROM live_rooms"
+    pool = await get_aiomysql_instance()
+    sql_data = await pool.query(sql)
+    bilibili_live_rooms = {}
+    for sd in sql_data:
+        bilibili_live_rooms[sd["room_id"]] = sd["username"]
+    return bilibili_live_rooms
 
 
 # 判断登录状态
-async def is_login(page):
+async def is_login(page) -> bool:
     await page.goto(bilibili_url)
-    if await page.get_by_text("修改资料").is_visible(timeout=5000):
-        bilibili_logger.info("登录成功")
+    if await page.get_by_text("每日奖励", exact=True).is_enabled():
+        login_res = bilibili_logger.info("Cookie 有效")
         return True
     else:
-        stderr = "Cookie 过期"
-        bilibili_logger.error(stderr)
-        push_msg(bilibili_title, stderr, bilibili_img)
+        login_res = "Cookie 失效"
+        bilibili_logger.error(login_res)
     return False
 
 
 # 签到
-async def sign(page):
-    msg = []
+async def sign(page) -> list:
+    message = [f"签到时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"]
 
     if not await is_login(page):
-        return "bilibili_not_login"
+        return "Cookie 失效"
 
     await page.goto(bilibili_sign_url)
-    if await page.get_by_text("无法重复签到").is_visible():
-        sign_res = f"签到时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n签到状态：今日已签到"
+    if await page.get_by_text("今日已签到").is_enabled():
+        sign_res = f"签到状态：今日已签到"
         bilibili_logger.info(sign_res)
-        msg.append(sign_res)
-    elif await page.get_by_text('"code":0').is_visible():
-        sign_res = f"签到时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n签到状态：签到成功"
+    elif await page.get_by_text('"code":0').is_enabled():
+        sign_res = f"签到状态：签到成功"
         bilibili_logger.info(sign_res)
-        msg.append(sign_res)
     else:
-        sign_res = f"签到失败"
+        sign_res = f"签到状态：签到失败"
         bilibili_logger.error(sign_res)
-        push_msg(bilibili_title, sign_res, bilibili_img)
-        msg.append(sign_res)
+    message.append(sign_res)
+    return message
 
-    return "\n".join(msg)
+
+async def bilibili_sign():
+    async with async_playwright() as playwright:
+        browser = await playwright.firefox.launch(headless=browser_headless)
+        context = await browser.new_context(storage_state=bilibili_cookie_file)
+        # 签到
+        page = await context.new_page()
+        task = asyncio.create_task(sign(page))
+        await asyncio.gather(task)
+        await page.close()
+        # 退出
+        await browser.close()
+        # 推送
+        message = "\n".join(task.result())
+        push_msg(title=bilibili_title, message=message, img_url=bilibili_img)
 
 
 # 直播间打卡
-async def live(page):
-    msg = []
+async def live(page) -> list:
+    message = [f"打卡时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"]
+
+    if not await is_login(page):
+        return "Cookie 失效"
 
     bilibili_live_rooms = await get_bilibili_live_rooms_from_pi()
     for room_id in list(bilibili_live_rooms.keys()):
         await page.goto(f"{bilibili_live_url}/{str(room_id)}")
-        # 输入框选择 xpath 绝对定位
-        input_box = page.locator(
-            "xpath=/html/body/div[3]/main/div[1]/section[1]/div[3]/div[12]/div/div[2]/div[2]/textarea",
-        )
+        # 输入框
+        input_box = page.get_by_role("textbox", name="发个弹幕呗~")
         # 输入内容
         await input_box.fill("1")
         # 发送按钮
-        send_button = page.locator(
-            "xpath=/html/body/div[3]/main/div[1]/section[1]/div[3]/div[12]/div/div[3]/div/button/span"
-        )
-        if await send_button.is_visible():
-            # 点击发送按钮
+        send_button = page.get_by_role("button", name="发送")
+        if await send_button.is_enabled():
             await send_button.click()
-            room_msg = f"直播间：【{bilibili_live_rooms[room_id]}】打卡成功，打卡时间：【{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}】"
-            bilibili_logger.info(room_msg)
+            room_res = f"直播间：【{bilibili_live_rooms[room_id]}】打卡成功"
+            bilibili_logger.info(room_res)
         else:
-            room_msg = f"直播间：【{bilibili_live_rooms[room_id]}】打卡失败，打卡时间：【{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}】"
-            bilibili_logger.error(room_msg)
-        msg.append(room_msg)
+            room_res = f"直播间：【{bilibili_live_rooms[room_id]}】打卡失败】"
+            bilibili_logger.error(room_res)
+        message.append(room_res)
 
-    return "\n".join(msg)
+    return message
 
 
-async def bilibili_sign():
+async def bilibili_live():
     async with async_playwright() as playwright:
         browser = await playwright.firefox.launch(headless=browser_headless)
         context = await browser.new_context(storage_state=bilibili_cookie_file)
-        # 签到
-        sign_page = await context.new_page()
-        sign_task = asyncio.create_task(sign(sign_page))
-        await asyncio.gather(sign_task)
-        await sign_page.close()
         # 直播间打卡
-        live_page = await context.new_page()
-        live_task = asyncio.create_task(live(live_page))
-        await asyncio.gather(live_task)
-        await live_page.close()
+        page = await context.new_page()
+        task = asyncio.create_task(live(page))
+        await asyncio.gather(task)
+        await page.close()
         # 退出
         await browser.close()
         # 推送
-        sign_result = sign_task.result()
-        live_result = live_task.result()
-        # ""分割两个服务的消息
-        push_msg(
-            title=bilibili_title,
-            message="\n".join([sign_result, "", live_result]),
-            img_url=bilibili_img,
-        )
+        message = "\n".join(task.result())
+        push_msg(title=bilibili_title, message=message, img_url=bilibili_img)
 
 
 if __name__ == "__main__":
     asyncio.run(bilibili_sign())
+    asyncio.run(bilibili_live())

_requests/[no_use]restart_vits_server.py

@@ -1,62 +0,0 @@
-# coding:utf-8
-
-import os
-import re
-import requests
-
-from os.path import abspath, dirname, join
-
-# conf
-title = "Huggingface"
-session = requests.Session()
-
-# from utils import push_msg
-
-
-def restart_vits_server():
-    headers = {
-        "authority": "huggingface.co",
-        "origin": "https://huggingface.co",
-        "referer": "https://huggingface.co/login",
-        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36",
-    }
-    data = {
-        "username": os.environ.get("hf_username"),
-        "password": os.environ.get("hf_password"),
-    }
-    session.post("https://huggingface.co/login", headers=headers, data=data)
-    rsp = session.get(
-        "https://huggingface.co/spaces/evi0mo/vits-fastapi-server/settings"
-    )
-    csrf = re.findall(r"name=\"csrf\" value=\"(.*?)\">", rsp.text)[0]
-
-    res = "init"
-    restart_headers = {
-        "authority": "huggingface.co",
-        "origin": "https://huggingface.co",
-        "referer": "https://huggingface.co/spaces/evi0mo/vits-fastapi-server/settings",
-        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36",
-    }
-
-    restart_data = {"csrf": csrf}
-    session.post(
-        "https://huggingface.co/spaces/evi0mo/vits-fastapi-server/restart?factory=false",
-        headers=restart_headers,
-        data=restart_data,
-    )
-    # check
-    check = session.get(
-        "https://huggingface.co/api/spaces/evi0mo/vits-fastapi-server/runtime"
-    )
-    if check.json()["stage"] == "RUNNING_BUILDING":
-        res = "success"
-        vits_logger.info(res)
-    else:
-        res = "failed"
-        vits_logger.error(res)
-
-    # push_msg(title, res)
-
-
-if __name__ == "__main__":
-    restart_vits_server()

_requests/aliyundrive.py

@@ -8,11 +8,7 @@ from os.path import abspath, dirname, join
 
 # conf
 sys.path.append(join(dirname(abspath(__file__)), "../"))
-from utils import (
-    push_msg,
-    get_aliyundrive_refresh_token,
-    update_aliyundrive_access_token,
-)
+from utils import push_msg
 from __init__ import (
     aliyundrive_nickname_url,
     aliyundrive_signin_url,
@@ -22,6 +18,7 @@ from __init__ import (
     aliyundrive_logger,
     aliyundrive_img,
 )
+from option_mysql import get_aiomysql_instance
 
 # session
 headers = {"message-Type": "application/json"}
@@ -34,6 +31,26 @@ def get_refresh_token():
     pass
 
 
+async def get_aliyundrive_refresh_token():
+    sql = f"SELECT token FROM aliyundrive WHERE id=1"
+    pool = await get_aiomysql_instance()
+    data = await pool.query(sql)
+    if data:
+        aliyundrive_logger.info("获取阿里云盘 refresh_token 成功")
+        return data[0]["token"]
+    return False
+
+
+async def update_aliyundrive_access_token(access_token):
+    sql = f"UPDATE aliyundrive SET token=%s WHERE id=2"
+    pool = await get_aiomysql_instance()
+    data = await pool.update(sql, [access_token])
+    if data:
+        aliyundrive_logger.info("update aliyundrive access token")
+        return True
+    return False
+
+
 async def validate_access_token():
     # token只有三小时有效期，所以定时任务刷新间隔为三小时
     # 从树莓派获取 refresh_token
@@ -97,8 +114,8 @@ def sign():
         # 当天签到获得的奖励信息
         reward_name = current_sign_info["reward"]["name"]
         reward_description = current_sign_info["reward"]["description"]
-        curent_reward = f"{reward_name}{reward_description}"
-        # 是否已领取奖励，bool
+        curent_reward = f"{reward_name}，{reward_description}"
+        # 是否已领取奖励
         is_reward = current_sign_info["isReward"]
         return is_reward, signin_count, curent_reward
     return False

api.py

@@ -11,7 +11,7 @@ from apscheduler.schedulers.background import BackgroundScheduler
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from loguru import logger
 
-from _playwright.bilibili import bilibili_sign
+from _playwright.bilibili import bilibili_sign, bilibili_live
 from _playwright.v2ex import v2ex_sign
 from _playwright.vits import vits_sign
 from _playwright.tsdm import tsdm_sign
@@ -28,15 +28,18 @@ def run_scheduler(loop):
     # add task
     task = AsyncIOScheduler()
     task.configure(timezone=pytz.timezone("Asia/Shanghai"))
+    task.add_job(tick, "cron", hour="*/2", next_run_time=datetime.now(), coalesce=True)
     task.add_job(
-        tick,
+        bilibili_sign,
         "cron",
-        hour="*/2",
+        hour=6,
+        minute=1,
         next_run_time=datetime.now(),
-        coalesce=True
+        coalesce=True,
+        misfire_grace_time=60 * 60 * 12,
     )
     task.add_job(
-        bilibili_sign,
+        bilibili_live,
         "cron",
         hour=6,
         minute=1,
@@ -105,9 +108,7 @@ def run_scheduler(loop):
 
 
 def start_daemon_scheduler(loop):
-    t = threading.Thread(
-        target=run_scheduler, args=(loop,), name="schedule", daemon=True
-    )
+    t = threading.Thread(target=run_scheduler, args=(loop,), name="schedule", daemon=True)
     t.start()
 
 

utils.py

@@ -105,7 +105,7 @@ def push_msg(title: str = "无标题", message: str = "无内容", img_url: str
         "message": message,
         "priority": 10,
         "extras": {
-            "client::display": {"contentType": "text/markdown"},
+            "client::display": {"contentType": "text/plaintext"},
             "client::notification": {"bigImageUrl": img_url},
         },
     }
@@ -113,16 +113,6 @@ def push_msg(title: str = "无标题", message: str = "无内容", img_url: str
     return {"code": resp.status_code, "res": resp.content}
 
 
-async def get_bilibili_live_rooms_from_pi() -> dict:
-    sql = f"SELECT room_id, username FROM live_rooms"
-    pool = await get_aiomysql_instance()
-    data = await pool.query(sql)
-    bilibili_live_rooms = {}
-    for d in data:
-        bilibili_live_rooms[d["room_id"]] = d["username"]
-    return bilibili_live_rooms
-
-
 async def get_cookie_from_pi() -> dict:
     sql = f"SELECT cookie_name, cookie_value FROM cookie"
     pool = await get_aiomysql_instance()
@@ -134,26 +124,6 @@ async def get_cookie_from_pi() -> dict:
     return data[0]["cookie_name"]
 
 
-async def get_aliyundrive_refresh_token():
-    sql = f"SELECT token FROM aliyundrive WHERE id=1"
-    pool = await get_aiomysql_instance()
-    data = await pool.query(sql)
-    if data:
-        logger.info("获取阿里云盘 refresh_token 成功")
-        return data[0]["token"]
-    return False
-
-
-async def update_aliyundrive_access_token(access_token):
-    sql = f"UPDATE aliyundrive SET token=%s WHERE id=2"
-    pool = await get_aiomysql_instance()
-    data = await pool.update(sql, [access_token])
-    if data:
-        logger.info("update aliyundrive access token")
-        return True
-    return False
-
-
 async def read_logfile(log_file: str) -> str:
     async with aiofiles.open(log_file, mode="r", encoding="utf-8") as f:
         data = await f.read()
@@ -183,7 +153,4 @@ async def main():
 if __name__ == "__main__":
     logger.debug(f"DEBUG Mode: {DEBUG}")
     logger.debug(f"DIR_PATH: {DIR_PATH}")
-    logger.debug(f"get_pi_ip: {get_pi_ip(gotify_host)}")
-    logger.debug(f'get_domain: {get_domain("https://github.com")}')
-    logger.debug(f'push_msg: {push_msg("utils.py", "hello world")}')
     # asyncio.run(main())