new build

.gitignore

@@ -0,0 +1,16 @@
+# cache
+__pycache__/
+sync.ffs_db
+*.pyc
+
+# file
+cookie/
+!cookie/.gitkeep
+log/
+!log/.gitkeep
+
+# dev
+config/settings_dev.py
+
+# test
+test.py

Dockerfile

@@ -0,0 +1,40 @@
+# build cache
+FROM python:3.9-slim-bullseye as builder
+
+WORKDIR /tmp
+
+# depends
+RUN apt-get update -y
+
+# pip
+COPY ./requirements.txt /tmp/requirements.txt
+RUN pip install -r /tmp/requirements.txt
+
+# playwright
+RUN playwright install-deps
+
+# Set up a new user named "user" with user ID 1000
+RUN useradd -m -u 1000 user
+
+# Switch to the "user" user
+USER user
+
+# Set home to the user's home directory
+ENV HOME=/home/user \
+	PATH=/home/user/.local/bin:$PATH
+
+# Set the working directory to the user's home directory
+WORKDIR $HOME/app
+
+RUN mkdir $HOME/app/cookie $HOME/app/log $HOME/app/_playwright $HOME/app/config
+
+# Try and run pip command after setting the user with `USER user` to avoid permission issues with Python
+RUN playwright install chromium firefox
+
+# use cache build
+FROM builder
+
+# Copy the current directory contents into the container at $HOME/app setting the owner to the user
+COPY --chown=user . $HOME/app
+
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

__init__.py

@@ -0,0 +1,14 @@
+# 全局变量 DIR_PATH 不能在其他地方用同一个命名，会被替换
+import sys
+from os.path import abspath, dirname
+
+
+DEBUG = False
+if DEBUG:
+    DIR_PATH = abspath(dirname(__file__))
+    sys.path.append(DIR_PATH)
+    from config.settings_dev import *
+else:
+    DIR_PATH = "/home/user/app"
+    sys.path.append(DIR_PATH)
+    from config.settings_prod import *

_playwright/[no_use]aliyundrive.py

@@ -0,0 +1,79 @@
+# 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

@@ -0,0 +1,123 @@
+# coding:utf-8
+import sys
+import asyncio
+
+from datetime import datetime
+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 __init__ import (
+    browser_headless,
+    browser_proxy,
+    bilibili_url,
+    bilibili_sign_url,
+    bilibili_live_url,
+    bilibili_cookie_file,
+    bilibili_title,
+    bilibili_logger,
+)
+
+bilibili_live_rooms = get_bilibili_live_rooms_from_pi()
+
+
+# 判断登录状态
+async def is_login(page):
+    await page.goto(bilibili_url)
+    if await page.get_by_text("修改资料").is_visible(timeout=2000):
+        bilibili_logger.info("登录成功")
+        return True
+    else:
+        stderr = "Cookie 过期"
+        bilibili_logger.error(stderr)
+        push_msg(bilibili_title, stderr)
+    return False
+
+
+# 签到
+async def sign(page):
+    msg = []
+
+    if not await is_login(page):
+        return "bilibili_not_login"
+
+    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签到状态：今日已签到"
+        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签到状态：签到成功"
+        bilibili_logger.info(sign_res)
+        msg.append(sign_res)
+    else:
+        sign_res = f"签到失败"
+        bilibili_logger.error(sign_res)
+        push_msg(bilibili_title, sign_res)
+        msg.append(sign_res)
+
+    return "\n".join(msg)
+
+
+# 直播间打卡
+async def live(page):
+    msg = []
+
+    if not await is_login(page):
+        return "bilibili_not_login"
+
+    for room_id in list(bilibili_live_rooms.keys()):
+        await page.goto(f"{bilibili_live_url}/{str(room_id)}")
+        # 输入框选择 xpath 绝对定位
+        input_box = await page.locator(
+            "xpath=/html/body/div[3]/main/div[1]/section[1]/div[3]/div[12]/div/div[2]/div[2]/textarea",
+        )
+        # 输入内容
+        await input_box.fill("1")
+        # 发送按钮
+        send_button = await 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():
+            # 点击发送按钮
+            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)
+        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)
+
+    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)
+        # 签到
+        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()
+        # 退出
+        await browser.close()
+        # 推送
+        sign_result = sign_task.result()
+        live_result = live_task.result()
+        # ""分割两个服务的消息
+        push_msg(bilibili_title, "\n".join([sign_result, "", live_result]))
+
+
+if __name__ == "__main__":
+    asyncio.run(bilibili_sign())

_playwright/tsdm.py

@@ -0,0 +1,116 @@
+# coding:utf-8
+import sys
+import random
+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,
+    tsdm_url,
+    tsdm_sign_url,
+    tsdm_cliwork_url,
+    tsdm_plugin_id,
+    tsdm_minds,
+    tsdm_cookie_file,
+    tsdm_title,
+    tsdm_logger,
+)
+
+
+# 判断登录状态
+async def is_login(page):
+    await page.goto(tsdm_url)
+    if await page.get_by_text("用户组").is_visible():
+        tsdm_logger.info("登录成功")
+        return True
+    else:
+        stderr = "Cookie 过期"
+        tsdm_logger.error(stderr)
+        push_msg(tsdm_title, stderr)
+    return False
+
+
+# 签到
+async def sign(page):
+    msg = []
+
+    if not await is_login(page):
+        return "tsdm_not_login"
+
+    # 签到
+    await page.goto(tsdm_sign_url)
+    if await page.get_by_text("今天已签到").is_visible():
+        sign_res = "今天已签到"
+        tsdm_logger.info(sign_res)
+        msg.append(sign_res)
+    else:
+        mind = random.choice(list(tsdm_minds.values()))
+        tsdm_logger.info(f"签到心情选择：{mind}")
+        await page.locator(
+            "#ct > div > div.bm.mtn > div.bm_c > form > select"
+        ).select_option(label=mind)
+        await page.get_by_label("不想填写").check()
+        await page.get_by_text("点我签到!").click()
+        sign_res = "签到成功"
+        tsdm_logger.info(sign_res)
+        msg.append(sign_res)
+
+    return "\n".join(msg)
+
+
+async def cliwork(page):
+    msg = []
+
+    if not await is_login(page):
+        return "tsdm_not_login"
+
+    # 打工
+    await page.goto(tsdm_url)  # 切换回电脑版
+    await page.goto(tsdm_cliwork_url)
+    if await page.get_by_text("您需要等待").is_visible():
+        has_work = "今天已打工"
+        tsdm_logger.info(has_work)
+        msg.append(has_work)
+    else:
+        for pid in tsdm_plugin_id:
+            await page.locator(f"#np_advid{pid} > a > img").click()
+            await page.bring_to_front()  # 回到第一个标签
+        await page.locator("#workstart").click()  # 领取奖励
+        work_res = "打工成功"
+        tsdm_logger.info(work_res)
+        msg.append(work_res)
+
+    return "\n".join(msg)
+
+
+async def tsdm_sign():
+    async with async_playwright() as playwright:
+        browser = await playwright.chromium.launch(headless=browser_headless)
+        context = await browser.new_context(storage_state=tsdm_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()
+        # 打工
+        cliwork_page = await context.new_page()
+        cliwork_task = asyncio.create_task(cliwork(cliwork_page))
+        await asyncio.gather(cliwork_task)
+        await cliwork_page.close()
+        # 退出
+        await browser.close()
+        # 推送
+        sign_result = sign_task.result()
+        cliwork_result = cliwork_task.result()
+        # ""分割两个服务的消息
+        push_msg(tsdm_title, "\n".join([sign_result, "", cliwork_result]))
+
+
+if __name__ == "__main__":
+    asyncio.run(tsdm_sign())

_playwright/v2ex.py

@@ -0,0 +1,72 @@
+# 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,
+    v2ex_url,
+    v2ex_sign_url,
+    v2ex_cookie_file,
+    v2ex_title,
+    v2ex_logger,
+)
+
+
+# 判断登录状态
+async def is_login(page):
+    await page.goto(v2ex_url)
+    if await page.get_by_text("记事本").is_visible():
+        v2ex_logger.info("登录成功")
+        return True
+    else:
+        stderr = "Cookie 过期"
+        v2ex_logger.error(stderr)
+        push_msg(v2ex_title, stderr)
+    return False
+
+
+# 签到
+async def sign(page):
+    msg = []
+
+    if not await is_login(page):
+        return "v2ex_not_login"
+
+    await page.goto(v2ex_sign_url)
+    if await page.get_by_text("每日登录奖励已领取").is_visible():
+        sign_res = "今日铜币已领取"
+        v2ex_logger.info(sign_res)
+        msg.append(sign_res)
+    else:
+        await page.get_by_text("领取 X 铜币").click()
+        sign_res = "领取铜币成功"
+        v2ex_logger.info(sign_res)
+        msg.append(sign_res)
+
+    return "\n".join(msg)
+
+
+async def v2ex_sign():
+    async with async_playwright() as playwright:
+        browser = await playwright.chromium.launch(headless=browser_headless)
+        context = await browser.new_context(storage_state=v2ex_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()
+        # 退出
+        await browser.close()
+        # 推送
+        sign_result = sign_task.result()
+        push_msg(v2ex_title, "\n".join([sign_result]))
+
+
+if __name__ == "__main__":
+    asyncio.run(v2ex_sign())

_playwright/vits.py

@@ -0,0 +1,66 @@
+# 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,
+    vits_url,
+    vits_cookie_file,
+    vits_title,
+    vits_logger,
+)
+
+
+# 判断登录状态
+async def is_login(page):
+    await page.goto(vits_url)
+    if await page.get_by_text("Pause Space").is_visible():
+        vits_logger.info("登录成功")
+        return True
+    else:
+        stderr = "Cookie 过期"
+        vits_logger.error(stderr)
+        push_msg(vits_title, stderr)
+    return False
+
+
+# 签到
+async def sign(page):
+    if not await is_login(page):
+        return "vits_not_login"
+
+    if await page.get_by_role("button", name="Restart space").is_visible():
+        await page.get_by_role("button", name="Restart space").click()
+        sign_res = "restart vits server success"
+        vits_logger.info(sign_res)
+    else:
+        sign_res = "restart vits server failed"
+        vits_logger.error(sign_res)
+
+    return sign_res
+
+
+async def vits_sign():
+    async with async_playwright() as playwright:
+        browser = await playwright.chromium.launch(headless=browser_headless)
+        context = await browser.new_context(storage_state=vits_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()
+        # 退出
+        await browser.close()
+        # 推送
+        sign_result = sign_task.result()
+        push_msg(vits_title, sign_result)
+
+
+if __name__ == "__main__":
+    asyncio.run(vits_sign())

_playwright/wuaipojie.py

@@ -0,0 +1,60 @@
+# 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,
+    wuaipojie_url,
+    wuaipojie_sign_url,
+    wuaipojie_cookie_file,
+    wuaipojie_title,
+    wuaipojie_logger,
+)
+
+
+async def sign(page):
+    await page.goto(wuaipojie_url)
+
+    # 判断登录状态
+    if await page.get_by_text("提醒").is_enabled():
+        wuaipojie_logger.info("登录成功")
+    else:
+        msg = "Cookie 过期"
+        wuaipojie_logger.error(msg)
+        return
+
+    # 签到
+    if await page.get_by_text("请下期再来").is_enabled():
+        msg = "今天已签到"
+        wuaipojie_logger.info(msg)
+    else:
+        await page.goto(wuaipojie_sign_url)
+        msg = "签到成功"
+        wuaipojie_logger.info(msg)
+    return msg
+
+
+async def wuaipojie_sign():
+    async with async_playwright() as playwright:
+        browser = await playwright.chromium.launch(headless=browser_headless)
+        context = await browser.new_context(storage_state=wuaipojie_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()
+        # 退出
+        await browser.close()
+        # 推送
+        sign_result = sign_task.result()
+        push_msg(wuaipojie_title, "\n".join([sign_result]))
+
+
+if __name__ == "__main__":
+    asyncio.run(wuaipojie_sign())

_requests/[no_use]restart_vits_server.py

@@ -0,0 +1,62 @@
+# 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

@@ -0,0 +1,121 @@
+# 接口参考：https://github.com/Anonym-w/autoSigninAliyun/blob/main/autoSignin.js
+import sys
+import requests
+
+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 __init__ import (
+    aliyundrive_nickname_url,
+    aliyundrive_signin_url,
+    aliyundrive_access_token_url,
+    aliyundrive_reward_url,
+    aliyundrive_title,
+    aliyundrive_logger,
+)
+
+# session
+headers = {"message-Type": "application/json"}
+session = requests.Session()
+query_body = {}
+
+
+def get_refresh_token():
+    # local storage only work in browser
+    return False
+
+
+def validate_access_token():
+    # token只有三小时有效期，所以定时任务刷新间隔为三小时
+    # 从树莓派获取 refresh_token
+    refresh_token = get_aliyundrive_refresh_token()
+    # 查询结构体
+    query_body.update({"grant_type": "refresh_token", "refresh_token": refresh_token})
+    # validate
+    resp = session.post(aliyundrive_access_token_url, headers=headers, json=query_body)
+    if resp.status_code != 200:
+        msg = "阿里云 access_token 失效"
+        aliyundrive_logger.error(msg)
+        push_msg(aliyundrive_title, msg)
+    else:
+        aliyundrive_logger.info("阿里云 access_token 有效")
+        new_access_token = resp.json()["access_token"]
+        update_aliyundrive_access_token(new_access_token)
+        headers["Authorization"] = f"Bearer {new_access_token}"
+        return new_access_token
+    return False
+
+
+def get_aliyun_nickname():
+    nickname = False
+    # 获取用户名
+    user_res = session.post(aliyundrive_nickname_url, headers=headers, json=query_body)
+    if user_res.status_code == 200:
+        msg = f"获取用户名成功，stdout：{user_res.json()}"
+        nickname = user_res.json()["nick_name"]
+    else:
+        msg = f"获取用户名失败，stderr：{user_res.json()}"
+        aliyundrive_logger.error(msg)
+        push_msg(title=aliyundrive_title, message=msg)
+    return nickname
+
+
+def sign():
+    # 签到
+    sign_res = session.post(aliyundrive_signin_url, headers=headers, json=query_body)
+    if sign_res.status_code != 200:
+        sign_msg = f"阿里云盘签到失败，stderr：{sign_res.json()}"
+        aliyundrive_logger.error(sign_msg)
+        push_msg(title=aliyundrive_title, message=sign_msg)
+    else:
+        sign_msg = f"阿里云盘签到成功"
+        aliyundrive_logger.info(sign_msg)
+        sign_jsondata = sign_res.json()
+        # 已签到天数
+        signin_count = sign_jsondata["result"]["signInCount"]
+        # 当天签到信息，数组下标 -1
+        current_sign_info = sign_jsondata["result"]["signInLogs"][signin_count - 1]
+        # 当天签到获得的奖励信息
+        reward_name = current_sign_info["reward"]["name"]
+        reward_description = current_sign_info["reward"]["description"]
+        curent_reward = f"{reward_name}{reward_description}"
+        # 是否已领取奖励，bool
+        is_reward = current_sign_info["isReward"]
+        return is_reward, signin_count, curent_reward
+    return False
+
+
+# 领取奖励
+def get_reward(signin_count):
+    query_body.update({"signInDay": signin_count})
+    reward_res = session.post(aliyundrive_reward_url, headers=headers, json=query_body)
+    if reward_res.status_code == 200:
+        aliyundrive_logger.info("领取奖励成功")
+    else:
+        reward_msg = f"领取奖励失败，stderr：{reward_res.json()}"
+        aliyundrive_logger.error(reward_msg)
+        push_msg(title=aliyundrive_title, message=reward_msg)
+        return reward_msg
+    return False
+
+
+def aliyundrive_sign():
+    validate_access_token()
+    nickname = get_aliyun_nickname()
+    is_reward, signin_count, curent_reward = sign()
+    if not is_reward:
+        get_reward(signin_count)
+    message = f"{nickname} 签到成功，本月累计签到 {signin_count} 天\n本次签到获得：{curent_reward}"
+    aliyundrive_logger.info(message)
+    push_msg(title=aliyundrive_title, message=message)
+    return message
+
+
+if __name__ == "__main__":
+    aliyundrive_sign()

api.py

@@ -0,0 +1,65 @@
+"""
+https://github.com/agronholm/apscheduler/blob/3.x/examples/schedulers/asyncio_.py
+"""
+import asyncio
+from collections.abc import Callable, Iterable, Mapping
+import threading
+from typing import Any
+import pytz
+
+from datetime import datetime
+from apscheduler.schedulers.background import BackgroundScheduler
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from loguru import logger
+
+from _playwright.bilibili import bilibili_sign
+from _playwright.v2ex import v2ex_sign
+from _playwright.vits import vits_sign
+from _playwright.tsdm import tsdm_sign
+from _playwright.wuaipojie import wuaipojie_sign
+from _requests.aliyundrive import aliyundrive_sign
+from utils import get_cookie_from_pi
+
+
+def tick():
+    logger.debug(f"Tick! The time is: {datetime.now():%Y-%m-%d %H:%M:%S}")
+
+
+def run_scheduler(loop):
+    asyncio.set_event_loop(loop)
+    # pi
+    get_cookie_from_pi()
+    # add task
+    sign_scheduler = AsyncIOScheduler()
+    sign_scheduler.configure(timezone=pytz.timezone("Asia/Shanghai"))
+    sign_scheduler.add_job(tick, "cron", hour="*/1")
+    sign_scheduler.add_job(bilibili_sign, "cron", hour=5, minute=1, second=0)
+    sign_scheduler.add_job(v2ex_sign, "cron", hour=5, minute=2, second=0)
+    sign_scheduler.add_job(tsdm_sign, "cron", hour=5, minute=3, second=0)
+    sign_scheduler.add_job(wuaipojie_sign, "cron", hour=5, minute=4, second=0)
+    sign_scheduler.add_job(vits_sign, "cron", hour=5, minute=5, second=0)
+    sign_scheduler.add_job(aliyundrive_sign, "cron", hour="*/2")
+    sign_scheduler.start()
+    try:
+        asyncio.get_event_loop().run_forever()
+    except (KeyboardInterrupt, SystemExit):
+        pass
+
+
+def start_daemon_scheduler(loop):
+    t = threading.Thread(target=run_scheduler, args=(loop,), name="schedule", daemon=True)
+    t.start()
+
+
+async def kill_daemon_scheduler():
+    loop = asyncio._get_running_loop()
+    loop.stop()
+
+
+if __name__ == "__main__":
+    schedule_loop = asyncio.new_event_loop()
+    start_daemon_scheduler(schedule_loop)
+    import time
+
+    time.sleep(3)
+    asyncio.run_coroutine_threadsafe(kill_daemon_scheduler(), loop=schedule_loop)

app.py

@@ -0,0 +1,100 @@
+# coding: utf-8
+import asyncio
+import os
+import requests
+
+from os.path import join
+from uvicorn import run
+from fastapi import FastAPI
+from fastapi.responses import Response
+from loguru import logger
+
+from __init__ import DIR_PATH, api_token
+from utils import push_msg
+from api import start_daemon_scheduler, kill_daemon_scheduler
+
+schedule_loop = asyncio.new_event_loop()
+
+app = FastAPI()
+
+
+@app.get("/")
+def home():
+    html_content = open("index.html").read()
+    return Response(content=html_content, status_code=200)
+
+
+# 开启定时任务
+@app.get("/run")
+def do_run(token: str):
+    if token != api_token:
+        return {"code": 422}
+
+    start_daemon_scheduler(schedule_loop)
+    res = {"code": 0, "data": "start scheduler"}
+    return res
+
+
+# 开启定时任务
+@app.get("/kill")
+def do_kill(token: str):
+    if token != api_token:
+        return {"code": 422}
+
+    asyncio.run_coroutine_threadsafe(kill_daemon_scheduler(), loop=schedule_loop)
+    res = {"code": 0, "data": "kill scheduler"}
+    return res
+
+
+# 查询日志
+@app.get("/log")
+async def do_log(token: str, name: str):
+    if token != api_token:
+        return {"code": 422}
+
+    log_file = join(DIR_PATH, f"log/{name}.log")
+    with open(log_file, "r", encoding="utf-8") as f:
+        data = f.read()
+
+    res = {"code": 200, "data": data}
+    return res
+
+
+# 列出当前目录所有文件
+@app.get("/path")
+def do_path(token: str):
+    if token != api_token:
+        return {"code": 422}
+
+    data = []
+    for root, _, files in os.walk(DIR_PATH):
+        for file in files:
+            data.append(join(root, file))
+    res = {"code": 0, "data": data}
+    return res
+
+
+# 测试连通性
+@app.get("/ip")
+def do_ip(token: str):
+    if token != api_token:
+        return {"code": 422}
+
+    ip = requests.get("https://checkip.amazonaws.com").text.strip()
+    logger.debug(f"checking ip: {ip}")
+    res = {"code": 0, "ip": ip}
+    return res
+
+
+# 测试推送消息
+@app.get("/push")
+def do_push(token: str):
+    if token != api_token:
+        return {"code": 422}
+
+    data = push_msg("send from huggingface docker", "hello world")
+    return {"code": 0, "data": data}
+
+
+if __name__ == "__main__":
+    run("app:app", host="0.0.0.0", port=8000, reload=True)

config/settings_prod.py

@@ -0,0 +1,108 @@
+# coding: utf-8
+import os
+import sys
+from os.path import join
+
+# 添加根目录引用
+# 不要用同一个 DIR_PATH 命名，会影响 __init__ 中的全局变量
+PROD_PATH = "/home/user/app"
+sys.path.append(PROD_PATH)
+
+from option_logger import register_logger
+
+# playwright
+browser_headless = True
+browser_proxy = {"server": None}
+user_agent = (
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36"
+)
+ignore_https_errors = True
+
+# api
+api_token = os.environ.get("token")
+
+# gotify
+gotify_host = os.environ.get("gotify_host")
+gotify_port = os.environ.get("gotify_port")
+gotify_token = os.environ.get("gotify_token")
+
+# mysql
+mysql_option = {
+    "HOST": os.environ.get("DB_HOST"),
+    "PORT": int(os.environ.get("DB_PORT")),
+    "USERNAME": os.environ.get("DB_USERNAME"),
+    "PASSWORD": os.environ.get("DB_PASSWORD"),
+    "DATABASE": os.environ.get("DB_DATABASE"),
+}
+
+# for fastapi
+logfile_mapping = {
+    "vits": join(PROD_PATH, "log/vits.log"),
+    "bilibili": join(PROD_PATH, "log/bilibili.log"),
+    "v2ex": join(PROD_PATH, "log/v2ex.log"),
+    "tsdm": join(PROD_PATH, "log/tsdm.log"),
+    "wuaipojie": join(PROD_PATH, "log/wuaipojie.log"),
+    "aliyundrive": join(PROD_PATH, "log/aliyundrive.log"),
+}
+
+# vits
+vits_title = "vits server"
+vits_url = "https://huggingface.co/spaces/evi0mo/vits-fastapi-server/settings"
+vits_logger = register_logger("vits", logfile_mapping["vits"])
+vits_cookie_file = join(PROD_PATH, "cookie/vits.json")
+hf_username = os.environ.get("hf_username")
+hf_password = os.environ.get("hf_password")
+
+# wuaipojie
+wuaipojie_title = "wuaipojie"
+wuaipojie_logger = register_logger("wuaipojie", logfile_mapping["wuaipojie"])
+wuaipojie_cookie_file = join(PROD_PATH, "cookie/wuaipojie.json")
+wuaipojie_url = "https://www.52pojie.cn/home.php?mod=task&do=apply&id=2"
+wuaipojie_sign_url = "https://www.52pojie.cn/home.php?mod=task&do=draw&id=2"
+
+# v2ex
+v2ex_title = "V2EX"
+v2ex_logger = register_logger("v2ex", logfile_mapping["v2ex"])
+v2ex_cookie_file = join(PROD_PATH, "cookie/v2ex.json")
+v2ex_url = "https://v2ex.com"
+v2ex_sign_url = "https://v2ex.com/mission/daily"
+
+# bilibili
+bilibili_title = "Bilibili"
+bilibili_logger = register_logger("bilibili", logfile_mapping["bilibili"])
+bilibili_cookie_file = join(PROD_PATH, "cookie/bilibili.json")
+bilibili_url = "https://account.bilibili.com/account/home"
+bilibili_sign_url = "https://api.live.bilibili.com/xlive/web-ucenter/v1/sign/DoSign"  # 签到地址
+bilibili_live_url = "https://live.bilibili.com"  # 直播间地址
+bilibili_live_rooms = os.environ.get("bilibili_live_rooms")
+
+# tsdm
+tsdm_title = "天使动漫"
+tsdm_logger = register_logger("tsdm", logfile_mapping["tsdm"])
+tsdm_cookie_file = join(PROD_PATH, "cookie/tsdm.json")
+tsdm_url = "https://www.tsdm39.com/forum.php?mobile=no"  # 首页
+tsdm_sign_url = "https://www.tsdm39.com/plugin.php?id=dsu_paulsign:sign&mobile=yes&simpletype=yes"  # 手机精简版签到地址
+tsdm_cliwork_url = "https://www.tsdm39.com/plugin.php?id=np_cliworkdz:work"  # 打工地址
+tsdm_plugin_id = [1, 2, 4, 6, 7, 9]  # 打工插件 id
+tsdm_minds = {
+    "kx": "开心",
+    "ng": "难过",
+    "ym": "郁闷",
+    "wl": "无聊",
+    "nu": "怒",
+    "ch": "擦汗",
+    "fd": "奋斗",
+    "yl": "慵懒",
+    "shuai": "衰",
+}  # 心情
+
+# aliyundrive
+aliyundrive_title = "阿里云盘"
+aliyundrive_logger = register_logger("aliyundrive", logfile_mapping["aliyundrive"])
+aliyundrive_refresh_token = os.environ.get("aliyundrive_refresh_token")
+aliyundrive_cookie_file = join(PROD_PATH, "cookie/aliyundrive.json")
+aliyundrive_url = "https://www.aliyundrive.com/drive"
+aliyundrive_nickname_url = "https://api.aliyundrive.com/adrive/v2/user/get"
+aliyundrive_access_token_url = "https://auth.aliyundrive.com/v2/account/token"
+aliyundrive_signin_url = "https://member.aliyundrive.com/v1/activity/sign_in_list"
+aliyundrive_reward_url = "https://member.aliyundrive.com/v1/activity/sign_in_reward"

index.html

@@ -0,0 +1,57 @@
+<html>
+<!-- Title -->
+
+<head>
+    <title>FastAPI Hello World</title>
+</head>
+
+<!-- Stylesheet -->
+<style>
+    body {
+        font-family: Arial, Helvetica, sans-serif;
+        font-size: 16px;
+        line-height: 1.5;
+        margin: 0;
+        padding: 0;
+    }
+
+    h1 {
+        font-size: 2em;
+        margin: 0;
+        padding: 0;
+        /* Center */
+        text-align: center;
+    }
+
+    h3 {
+        margin: 0;
+        padding: 0;
+        /* Center */
+        text-align: center;
+    }
+</style>
+
+<!-- Body -->
+
+<body>
+    <h1>FastAPI Hello World Example</h1>
+
+    <h3>
+        This is a simple example of a FastAPI Hello World application using Docker. See the links below for more
+        information.
+    </h3>
+    <!-- List of different relevant links and descriptions about them -->
+
+    <div style="margin: 0 auto; width: 50%;">
+
+        <ul>
+            <!-- Link to self /docs -->
+            <li>
+                <a href="/docs">This API's Interactive Swaggar Docs</a> - Try the API out here!
+            </li>
+        </ul>
+    </div>
+
+</body>
+
+</html>

option_logger.py

@@ -0,0 +1,12 @@
+from loguru import logger
+
+
+# 根据名称注册多个 logger
+def register_logger(bind_name, file_path):
+    logger.add(
+        file_path,
+        format="{time:YYYY-MM-DD HH:mm:ss} {level} {message}",
+        filter=lambda record: record["extra"].get("name") == bind_name,
+    )
+    new_logger = logger.bind(name=bind_name)
+    return new_logger

option_mysql.py

@@ -0,0 +1,54 @@
+import pymysql
+
+
+class OptionMysql(object):
+    def __init__(self, options):
+        host = options["HOST"]
+        user = options["USERNAME"]
+        password = options["PASSWORD"]
+        database = options["DATABASE"]
+        port = options["PORT"]
+        charset = "utf8"
+        # 连接数据库
+        self.conn = pymysql.connect(
+            host=host,
+            port=port,
+            user=user,
+            password=password,
+            database=database,
+            charset=charset,
+        )
+        # 创建游标
+        self.cur = self.conn.cursor()
+        self.dict_cur = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
+
+    def __exit__(self):
+        self.cur.close()
+        self.dict_cur.close()
+        self.conn.close()
+
+    def get_dict_data(self, sql, params=[]):
+        """查询，返回字典类型"""
+        try:
+            if params:
+                self.dict_cur.execute(sql, params)
+            else:
+                self.dict_cur.execute(sql)
+            data = self.dict_cur.fetchall()
+            return data
+        except Exception as e:
+            self.conn.rollback()
+            raise e
+
+    def update_data(self, sql, params=[]):
+        """更新"""
+        try:
+            if not params:
+                self.cur.execute(sql)
+            else:
+                self.cur.execute(sql, params)
+            self.conn.commit()
+        except Exception as e:
+            self.conn.rollback()
+            raise e
+        return True

requirements.txt

@@ -0,0 +1,15 @@
+apscheduler
+psutil
+gotify
+loguru
+playwright
+playwright-stealth
+requests
+tld
+uvicorn
+fastapi
+pytz
+pymysql
+#ddddocr
+#Pillow
+#selenium

utils.py

@@ -0,0 +1,181 @@
+# coding:utf-8
+import os
+import sys
+import socket
+import asyncio
+import psutil
+import requests
+
+from os.path import join
+from tld import get_tld
+from loguru import logger
+from playwright.async_api import async_playwright, Page
+from playwright_stealth import stealth_async
+
+from option_mysql import OptionMysql
+from __init__ import (
+    gotify_host,
+    gotify_port,
+    gotify_token,
+    mysql_option,
+    DIR_PATH,
+    DEBUG,
+    browser_headless,
+    browser_proxy,
+    ignore_https_errors,
+    user_agent,
+    aliyundrive_access_token_url,
+)
+
+
+# webdriver
+async def init_page(storage_state=None, browser_type: str = "chromium") -> Page:
+    if browser_type not in ("chromium", "firefox", "webkit"):
+        raise TypeError("unspported browser")
+
+    async with async_playwright() as p:
+        if browser_type == "chromium":
+            browser = await p.chromium.launch(headless=browser_headless)
+        elif browser_type == "firefox":
+            browser = await p.firefox.launch(headless=browser_headless)
+        else:
+            logger.error("初始化浏览器失败")
+            exit(-1)
+
+        context = await browser.new_context(
+            proxy=browser_proxy,
+            ignore_https_errors=ignore_https_errors,
+            user_agent=user_agent,
+        )
+        # 从文件读取 Cookie
+        context = await browser.new_context(storage_state=storage_state)
+        page = await context.new_page()
+        # 隐藏 webdriver 特征
+        await stealth_async(page)
+        # await page.pause()  # async 跳出上下文就会 playwright._impl._api_types.Error: Connection closed
+        await goto_github(page)
+        await dump_cookie(page)  # 创建函数：一个 browser 对应一个 context、page，在 with 内完成多个任务
+        await browser.close()
+
+
+async def goto_github(page):
+    await page.goto("https://github.com")
+
+
+async def dump_cookie(page):
+    # url = "https://www.52pojie.cn/home.php"
+    # url = "https://v2ex.com"
+    # url = "https://www.bilibili.com"
+    # url = "https://www.tsdm39.com/forum.php?mobile=no"
+    # url = "https://www.aliyundrive.com/drive"
+    url = "https://huggingface.co/login"
+    save_cookie = join(DIR_PATH, f"{get_domain(url)}.json")
+    await page.goto(url)
+    logger.debug(DIR_PATH)
+    await page.screenshot(
+        path=join(DIR_PATH, "screenshot.png"),
+        full_page=True,
+    )  # 截图
+    while input("input c to continue: ") != "c":  # 阻塞等待填写登录信息
+        continue
+    await page.context.storage_state(path=save_cookie)  # 保存 Cookie 到文件
+
+
+# 获取容器内 IP
+def get_local_ip():
+    net_card_info = []
+    info = psutil.net_if_addrs()
+    for k, v in info.items():
+        for item in v:
+            if item[0] == 2 and not item[1] == "127.0.0.1":
+                net_card_info.append(item[1])
+    local_ip = net_card_info[0]
+    return local_ip
+
+
+# 从 url 获取域名
+def get_domain(url: str) -> str:
+    result = get_tld(url, as_object=True)
+    domain = result.domain
+    return domain
+
+
+# 从域名获取 IP
+def get_domain_ip(host: str) -> str:
+    domain_ip = socket.gethostbyname(host)
+    return domain_ip
+
+
+# gotify 推送
+def push_msg(title, message):
+    gotify_ip = get_domain_ip(gotify_host)
+    url = f"http://{gotify_ip}:{gotify_port}/message?token={gotify_token}"
+    data = {"title": title, "message": message, "priority": 10}
+    with requests.post(url, json=data) as resp:
+        return resp.json()
+
+
+def get_bilibili_live_rooms_from_pi() -> dict:
+    logger.info("getting bilibili_live_rooms from pi")
+    bilibili_live_rooms = {}
+    mysql = OptionMysql(mysql_option)
+    sql = f"SELECT room_id, username FROM live_rooms"
+    data = mysql.get_dict_data(sql)
+    if data:
+        for d in data:
+            bilibili_live_rooms[d["room_id"]] = d["username"]
+    return bilibili_live_rooms
+
+
+def get_cookie_from_pi() -> dict:
+    logger.info("getting cookie from pi")
+    mysql = OptionMysql(mysql_option)
+    sql = f"SELECT cookie_name, cookie_value FROM cookie"
+    data = mysql.get_dict_data(sql)
+    for d in data:
+        log_file = join(DIR_PATH, f"cookie/{d['cookie_name']}.json")
+        with open(log_file, "w") as lf:
+            lf.write(d["cookie_value"])
+    return data[0]["cookie_name"]
+
+
+def get_aliyundrive_refresh_token():
+    mysql = OptionMysql(mysql_option)
+    sql = f"SELECT token FROM aliyundrive WHERE id=1"
+    data = mysql.get_dict_data(sql)
+    if data:
+        logger.info("获取阿里云盘 refresh_token 成功")
+        return data[0]["token"]
+    return False
+
+
+def update_aliyundrive_access_token(access_token):
+    mysql = OptionMysql(mysql_option)
+    sql = f"UPDATE aliyundrive SET token=%s WHERE id=2"
+    data = mysql.update_data(sql, [access_token])
+    if data:
+        logger.info("update aliyundrive access token")
+        return True
+    return False
+
+
+def get_log(log_file: str) -> str:
+    with open(log_file, "r", encoding="utf-8") as f:
+        data = f.read()
+        return data
+
+
+if __name__ == "__main__":
+    logger.debug(f"DEBUG Mode: {DEBUG}")
+    logger.debug(f"DIR_PATH: {DIR_PATH}")
+    # asyncio.run(init_page())
+    # logger.debug(f"get_domain_ip: {get_domain_ip(gotify_host)}")
+    # logger.debug(f'get_domain: {get_domain("https://github.com")}')
+    # logger.debug(f"list_path: {list_path()}")
+    # logger.debug(f'push_msg: {push_msg("utils.py", "hello world")}')
+    # logger.debug(f"get_aliyun_token_from_pi(): {get_aliyun_token_from_pi()}")
+    # logger.debug(f"get_bilibili_live_rooms_from_pi(): {get_bilibili_live_rooms_from_pi()}")
+    # logger.debug(f"get_cookie_from_pi(): {get_cookie_from_pi()}")
+    # logger.debug(f'get_cron_log(): {get_cron_log("log/tsdm.log")}')
+    # logger.debug(f"get_aliyundrive_access_token: {get_aliyundrive_access_token()}")
+    # logger.debug(f"update_aliyundrive_access_token: {update_aliyundrive_access_token('123')}")