diff --git a/package.v2.json b/package.v2.json index 5ea2dc6..363080b 100644 --- a/package.v2.json +++ b/package.v2.json @@ -1,4 +1,22 @@ { + "AutoSignIn": { + "name": "站点自动签到", + "description": "自动模拟登录、签到站点。", + "labels": "站点", + "version": "2.6.1", + "icon": "signin.png", + "author": "thsrite", + "level": 2, + "history": { + "v2.6.1": "修复历史记录只显示一天!", + "v2.6": "感谢madrays佬提供的UI!", + "v2.5.4": "增加保号风险提示", + "v2.5.3": "优化执行周期输入，需要MoviePilot v2.2.1+", + "v2.5.2": "修复HDArea签到", + "v2.5.1": "修复空签到失败问题", + "v2.5": "MoviePilot V2 版本站点自动签到插件" + } + }, "EmbyMetaRefresh": { "name": "Emby元数据刷新", "description": "定时刷新Emby媒体库元数据，演职人员中文。", diff --git a/plugins.v2/autosignin/__init__.py b/plugins.v2/autosignin/__init__.py new file mode 100644 index 0000000..c28b84b --- /dev/null +++ b/plugins.v2/autosignin/__init__.py @@ -0,0 +1,1811 @@ +import re +import traceback +from datetime import datetime, timedelta +from multiprocessing.dummy import Pool as ThreadPool +from multiprocessing.pool import ThreadPool +from typing import Any, List, Dict, Tuple, Optional +from urllib.parse import urljoin + +import pytz +from apscheduler.schedulers.background import BackgroundScheduler +from apscheduler.triggers.cron import CronTrigger +from ruamel.yaml import CommentedMap + +from app import schemas +from app.chain.site import SiteChain +from app.core.config import settings +from app.core.event import EventManager, eventmanager, Event +from app.db.site_oper import SiteOper +from app.helper.browser import PlaywrightHelper +from app.helper.cloudflare import under_challenge +from app.helper.module import ModuleHelper +from app.helper.sites import SitesHelper +from app.log import logger +from app.plugins import _PluginBase +from app.schemas.types import EventType, NotificationType +from app.utils.http import RequestUtils +from app.utils.site import SiteUtils +from app.utils.string import StringUtils +from app.utils.timer import TimerUtils + + +class AutoSignIn(_PluginBase): + # 插件名称 + plugin_name = "站点自动签到" + # 插件描述 + plugin_desc = "自动模拟登录、签到站点。" + # 插件图标 + plugin_icon = "signin.png" + # 插件版本 + plugin_version = "2.6.1" + # 插件作者 + plugin_author = "thsrite" + # 作者主页 + author_url = "https://github.com/thsrite" + # 插件配置项ID前缀 + plugin_config_prefix = "autosignin_" + # 加载顺序 + plugin_order = 0 + # 可使用的用户级别 + auth_level = 2 + + # 私有属性 + sites: SitesHelper = None + siteoper: SiteOper = None + sitechain: SiteChain = None + # 事件管理器 + event: EventManager = None + # 定时器 + _scheduler: Optional[BackgroundScheduler] = None + # 加载的模块 + _site_schema: list = [] + + # 配置属性 + _enabled: bool = False + _cron: str = "" + _onlyonce: bool = False + _notify: bool = False + _queue_cnt: int = 5 + _sign_sites: list = [] + _login_sites: list = [] + _retry_keyword = None + _clean: bool = False + _start_time: int = None + _end_time: int = None + _auto_cf: int = 0 + + def init_plugin(self, config: dict = None): + self.sites = SitesHelper() + self.siteoper = SiteOper() + self.event = EventManager() + self.sitechain = SiteChain() + + # 停止现有任务 + self.stop_service() + + # 配置 + if config: + self._enabled = config.get("enabled") + self._cron = config.get("cron") + self._onlyonce = config.get("onlyonce") + self._notify = config.get("notify") + self._queue_cnt = config.get("queue_cnt") or 5 + self._sign_sites = config.get("sign_sites") or [] + self._login_sites = config.get("login_sites") or [] + self._retry_keyword = config.get("retry_keyword") + self._auto_cf = config.get("auto_cf") + self._clean = config.get("clean") + + # 过滤掉已删除的站点 + all_sites = [site.id for site in self.siteoper.list_order_by_pri()] + [site.get("id") for site in + self.__custom_sites()] + self._sign_sites = [site_id for site_id in all_sites if site_id in self._sign_sites] + self._login_sites = [site_id for site_id in all_sites if site_id in self._login_sites] + # 保存配置 + self.__update_config() + + # 加载模块 + if self._enabled or self._onlyonce: + + self._site_schema = ModuleHelper.load('app.plugins.autosignin.sites', + filter_func=lambda _, obj: hasattr(obj, 'match')) + + # 立即运行一次 + if self._onlyonce: + # 定时服务 + self._scheduler = BackgroundScheduler(timezone=settings.TZ) + logger.info("站点自动签到服务启动，立即运行一次") + self._scheduler.add_job(func=self.sign_in, trigger='date', + run_date=datetime.now(tz=pytz.timezone(settings.TZ)) + timedelta(seconds=3), + name="站点自动签到") + + # 关闭一次性开关 + self._onlyonce = False + # 保存配置 + self.__update_config() + + # 启动任务 + if self._scheduler.get_jobs(): + self._scheduler.print_jobs() + self._scheduler.start() + + def get_state(self) -> bool: + return self._enabled + + def __update_config(self): + # 保存配置 + self.update_config( + { + "enabled": self._enabled, + "notify": self._notify, + "cron": self._cron, + "onlyonce": self._onlyonce, + "queue_cnt": self._queue_cnt, + "sign_sites": self._sign_sites, + "login_sites": self._login_sites, + "retry_keyword": self._retry_keyword, + "auto_cf": self._auto_cf, + "clean": self._clean, + } + ) + + @staticmethod + def get_command() -> List[Dict[str, Any]]: + """ + 定义远程控制命令 + :return: 命令关键字、事件、描述、附带数据 + """ + return [{ + "cmd": "/site_signin", + "event": EventType.PluginAction, + "desc": "站点签到", + "category": "站点", + "data": { + "action": "site_signin" + } + }] + + def get_api(self) -> List[Dict[str, Any]]: + """ + 获取插件API + [{ + "path": "/xx", + "endpoint": self.xxx, + "methods": ["GET", "POST"], + "summary": "API说明" + }] + """ + return [{ + "path": "/signin_by_domain", + "endpoint": self.signin_by_domain, + "methods": ["GET"], + "summary": "站点签到", + "description": "使用站点域名签到站点", + }] + + def get_service(self) -> List[Dict[str, Any]]: + """ + 注册插件公共服务 + [{ + "id": "服务ID", + "name": "服务名称", + "trigger": "触发器：cron/interval/date/CronTrigger.from_crontab()", + "func": self.xxx, + "kwargs": {} # 定时器参数 + }] + """ + if self._enabled and self._cron: + try: + if str(self._cron).strip().count(" ") == 4: + return [{ + "id": "AutoSignIn", + "name": "站点自动签到服务", + "trigger": CronTrigger.from_crontab(self._cron), + "func": self.sign_in, + "kwargs": {} + }] + else: + # 2.3/9-23 + crons = str(self._cron).strip().split("/") + if len(crons) == 2: + # 2.3 + cron = crons[0] + # 9-23 + times = crons[1].split("-") + if len(times) == 2: + # 9 + self._start_time = int(times[0]) + # 23 + self._end_time = int(times[1]) + if self._start_time and self._end_time: + return [{ + "id": "AutoSignIn", + "name": "站点自动签到服务", + "trigger": "interval", + "func": self.sign_in, + "kwargs": { + "hours": float(str(cron).strip()), + } + }] + else: + logger.error("站点自动签到服务启动失败，周期格式错误") + else: + # 默认0-24 按照周期运行 + return [{ + "id": "AutoSignIn", + "name": "站点自动签到服务", + "trigger": "interval", + "func": self.sign_in, + "kwargs": { + "hours": float(str(self._cron).strip()), + } + }] + except Exception as err: + logger.error(f"定时任务配置错误：{str(err)}") + elif self._enabled: + # 随机时间 + triggers = TimerUtils.random_scheduler(num_executions=2, + begin_hour=9, + end_hour=23, + max_interval=6 * 60, + min_interval=2 * 60) + ret_jobs = [] + for trigger in triggers: + ret_jobs.append({ + "id": f"AutoSignIn|{trigger.hour}:{trigger.minute}", + "name": "站点自动签到服务", + "trigger": "cron", + "func": self.sign_in, + "kwargs": { + "hour": trigger.hour, + "minute": trigger.minute + } + }) + return ret_jobs + return [] + + def get_form(self) -> Tuple[List[dict], Dict[str, Any]]: + """ + 拼装插件配置页面，需要返回两块数据：1、页面配置；2、数据结构 + """ + # 站点的可选项（内置站点 + 自定义站点） + customSites = self.__custom_sites() + + site_options = ([{"title": site.name, "value": site.id} + for site in self.siteoper.list_order_by_pri()] + + [{"title": site.get("name"), "value": site.get("id")} + for site in customSites]) + return [ + { + 'component': 'VForm', + 'content': [ + { + 'component': 'VRow', + 'content': [ + { + 'component': 'VCol', + 'props': { + 'cols': 12, + 'md': 3 + }, + 'content': [ + { + 'component': 'VSwitch', + 'props': { + 'model': 'enabled', + 'label': '启用插件', + } + } + ] + }, + { + 'component': 'VCol', + 'props': { + 'cols': 12, + 'md': 3 + }, + 'content': [ + { + 'component': 'VSwitch', + 'props': { + 'model': 'notify', + 'label': '发送通知', + } + } + ] + }, + { + 'component': 'VCol', + 'props': { + 'cols': 12, + 'md': 3 + }, + 'content': [ + { + 'component': 'VSwitch', + 'props': { + 'model': 'onlyonce', + 'label': '立即运行一次', + } + } + ] + }, + { + 'component': 'VCol', + 'props': { + 'cols': 12, + 'md': 3 + }, + 'content': [ + { + 'component': 'VSwitch', + 'props': { + 'model': 'clean', + 'label': '清理本日缓存', + } + } + ] + } + ] + }, + { + 'component': 'VRow', + 'content': [ + { + 'component': 'VCol', + 'props': { + 'cols': 12, + 'md': 6 + }, + 'content': [ + { + 'component': 'VCronField', + 'props': { + 'model': 'cron', + 'label': '执行周期', + 'placeholder': '5位cron表达式，留空自动' + } + } + ] + }, + { + 'component': 'VCol', + 'props': { + 'cols': 12, + 'md': 6 + }, + 'content': [ + { + 'component': 'VTextField', + 'props': { + 'model': 'queue_cnt', + 'label': '队列数量' + } + } + ] + }, + { + 'component': 'VCol', + 'props': { + 'cols': 12, + 'md': 6 + }, + 'content': [ + { + 'component': 'VTextField', + 'props': { + 'model': 'retry_keyword', + 'label': '重试关键词', + 'placeholder': '支持正则表达式，命中才重签' + } + } + ] + }, + { + 'component': 'VCol', + 'props': { + 'cols': 12, + 'md': 6 + }, + 'content': [ + { + 'component': 'VTextField', + 'props': { + 'model': 'auto_cf', + 'label': '自动优选', + 'placeholder': '命中重试关键词次数（0-关闭）' + } + } + ] + } + ] + }, + { + 'component': 'VRow', + 'content': [ + { + 'component': 'VCol', + 'content': [ + { + 'component': 'VSelect', + 'props': { + 'chips': True, + 'multiple': True, + 'model': 'sign_sites', + 'label': '签到站点', + 'items': site_options + } + } + ] + } + ] + }, + { + 'component': 'VRow', + 'content': [ + { + 'component': 'VCol', + 'content': [ + { + 'component': 'VSelect', + 'props': { + 'chips': True, + 'multiple': True, + 'model': 'login_sites', + 'label': '登录站点', + 'items': site_options + } + } + ] + } + ] + }, + { + 'component': 'VRow', + 'content': [ + { + 'component': 'VCol', + 'props': { + 'cols': 12, + }, + 'content': [ + { + 'component': 'VAlert', + 'props': { + 'type': 'info', + 'variant': 'tonal', + 'text': '执行周期支持：' + '1、5位cron表达式；' + '2、配置间隔（小时），如2.3/9-23（9-23点之间每隔2.3小时执行一次）；' + '3、周期不填默认9-23点随机执行2次。' + '每天首次全量执行，其余执行命中重试关键词的站点。' + } + } + ] + } + ] + }, + { + 'component': 'VRow', + 'content': [ + { + 'component': 'VCol', + 'props': { + 'cols': 12, + }, + 'content': [ + { + 'component': 'VAlert', + 'props': { + 'type': 'info', + 'variant': 'tonal', + 'text': '自动优选：0-关闭，命中重试关键词次数大于该数量时自动执行Cloudflare IP优选（需要开启且则正确配置Cloudflare IP优选插件和自定义Hosts插件）' + } + } + ] + } + ] + }, + { + 'component': 'VRow', + 'content': [ + { + 'component': 'VCol', + 'props': { + 'cols': 12, + }, + 'content': [ + { + 'component': 'VAlert', + 'props': { + 'type': 'warning', + 'variant': 'tonal', + 'text': '不是所有的站点都会把程序自动登录/签到定义为用户活跃（比如馒头），提示签到/登录成功仍然存在掉号风险！请结合站点公告说明自行把握。' + } + } + ] + } + ] + } + ] + } + ], { + "enabled": False, + "notify": True, + "cron": "", + "auto_cf": 0, + "onlyonce": False, + "clean": False, + "queue_cnt": 5, + "sign_sites": [], + "login_sites": [], + "retry_keyword": "错误|失败" + } + + def __custom_sites(self) -> List[Any]: + custom_sites = [] + custom_sites_config = self.get_config("CustomSites") + if custom_sites_config and custom_sites_config.get("enabled"): + custom_sites = custom_sites_config.get("sites") + return custom_sites + + def get_page(self) -> List[dict]: + """ + 拼装插件详情页面，需要返回页面配置，同时附带数据 + """ + # 获取最近14天的日期数组 + date_list = [(datetime.now() - timedelta(days=i)).date() for i in range(14)] + + # 获取所有数据，包括签到和登录历史 + all_data = { + "signin": [], # 签到数据 + "login": [] # 登录数据 + } + sign_dates = set() + sites_info = {} # 记录站点信息 + + # 获取站点信息 + site_indexers = self.sites.get_indexers() + for site in site_indexers: + if not site.get("public"): + sites_info[site.get("id")] = site.get("name") + + # 自定义站点 + custom_sites = self.__custom_sites() + for site in custom_sites: + sites_info[site.get("id")] = site.get("name") + + # 获取常规日期格式数据 + for day in date_list: + day_str = f"{day.month}月{day.day}日" + day_formatted = day.strftime('%Y-%m-%d') + + # 获取"月日"格式数据 + day_data = self.get_data(day_str) + if day_data: + # 添加日期信息到每条记录 + if isinstance(day_data, list): + for record in day_data: + if isinstance(record, dict): + record["date"] = day_str + record["day_obj"] = day + # 区分签到和登录数据 + if "登录" in record.get("status", ""): + all_data["login"].append(record) + else: + all_data["signin"].append(record) + sign_dates.add(day_str) + + # 获取"签到-yyyy-mm-dd"和"登录-yyyy-mm-dd"格式数据 + signin_history = self.get_data(key="签到-" + day_formatted) + if signin_history: + if isinstance(signin_history, dict): + # 获取完成签到的站点ID列表 + done_sites = signin_history.get("do", []) + retry_sites = signin_history.get("retry", []) + + # 为所有已完成签到的站点创建记录 + for site_id in done_sites: + site_id_str = str(site_id) + site_name = sites_info.get(site_id_str) or sites_info.get(site_id) or f"站点ID: {site_id}" + + # 跳过需要重试的站点 + if site_id in retry_sites: + # 为需要重试的站点添加记录 + status_text = "需要重试" + all_data["signin"].append({ + "site": site_name, + "status": status_text, + "date": day_str, + "day_obj": day, + "site_id": site_id + }) + else: + # 为已完成的站点添加记录 + status_text = "已签到" + all_data["signin"].append({ + "site": site_name, + "status": status_text, + "date": day_str, + "day_obj": day, + "site_id": site_id + }) + + sign_dates.add(day_str) + + # 获取登录历史数据 + login_history = self.get_data(key="登录-" + day_formatted) + if login_history: + if isinstance(login_history, dict): + # 获取完成登录的站点ID列表 + done_sites = login_history.get("do", []) + retry_sites = login_history.get("retry", []) + + # 为所有已完成登录的站点创建记录 + for site_id in done_sites: + site_id_str = str(site_id) + site_name = sites_info.get(site_id_str) or sites_info.get(site_id) or f"站点ID: {site_id}" + + # 跳过需要重试的站点 + if site_id in retry_sites: + # 为需要重试的站点添加记录 + status_text = "登录需要重试" + all_data["login"].append({ + "site": site_name, + "status": status_text, + "date": day_str, + "day_obj": day, + "site_id": site_id + }) + else: + # 为已完成的站点添加记录 + status_text = "登录成功" + all_data["login"].append({ + "site": site_name, + "status": status_text, + "date": day_str, + "day_obj": day, + "site_id": site_id + }) + + sign_dates.add(day_str) + + # 如果没有数据，显示提示信息 + if not all_data["signin"] and not all_data["login"]: + return [{ + 'component': 'VAlert', + 'props': { + 'type': 'info', + 'text': '暂无签到数据', + 'variant': 'tonal', + 'class': 'mt-4', + 'prepend-icon': 'mdi-information' + } + }] + + # 确保签到数据中至少有所有日期的记录 + if sign_dates: + sign_dates_list = list(sign_dates) + sign_dates_list.sort(reverse=True) # 最新日期优先 + else: + sign_dates_list = [f"{date_list[0].month}月{date_list[0].day}日"] + + # 按站点分组并去重数据 + signin_site_data = {} + login_site_data = {} + + # 处理签到数据 - 每个站点每天只保留一条最新记录 + site_day_records = {} # 用于去重: {site}_{date} -> record + for data in all_data["signin"]: + site_name = data.get("site", "未知站点") + date_str = data.get("date", "") + site_day_key = f"{site_name}_{date_str}" + + # 存储或更新记录（如有多条取最新） + site_day_records[site_day_key] = data + + # 整理去重后的数据 + for key, record in site_day_records.items(): + site_name = record.get("site", "未知站点") + if site_name not in signin_site_data: + signin_site_data[site_name] = [] + signin_site_data[site_name].append(record) + + # 处理登录数据 - 同样去重 + site_day_records = {} # 重置去重字典 + for data in all_data["login"]: + site_name = data.get("site", "未知站点") + date_str = data.get("date", "") + site_day_key = f"{site_name}_{date_str}" + + # 存储或更新记录 + site_day_records[site_day_key] = data + + # 整理去重后的数据 + for key, record in site_day_records.items(): + site_name = record.get("site", "未知站点") + if site_name not in login_site_data: + login_site_data[site_name] = [] + login_site_data[site_name].append(record) + + # 创建签到折叠面板 + signin_panels = [] + for site_name, records in signin_site_data.items(): + # 按日期排序，最新的在前面 + try: + records.sort(key=lambda x: x.get("day_obj", datetime.now().date()), reverse=True) + except: + pass # 排序失败时跳过 + + # 获取最新的状态作为站点概要 + latest_status = records[0].get("status", "未知状态") + + # 确定状态颜色和图标 + status_color = "teal-lighten-3" + status_icon = "mdi-emoticon-happy-outline" + + if "失败" in latest_status or "错误" in latest_status: + status_color = "deep-orange-lighten-3" + status_icon = "mdi-emoticon-sad-outline" + elif "Cookie已失效" in latest_status: + status_color = "pink-lighten-3" + status_icon = "mdi-cookie-off" + elif "重试" in latest_status: + status_color = "amber-lighten-3" + status_icon = "mdi-emoticon-confused-outline" + elif "已签到" in latest_status: + status_color = "light-blue-lighten-3" + status_icon = "mdi-emoticon-cool-outline" + elif "成功" in latest_status: + status_color = "teal-lighten-3" + status_icon = "mdi-emoticon-happy-outline" + + # 创建每个站点的折叠面板 + signin_panels.append( + self._create_expansion_panel(site_name, records, status_color, status_icon, latest_status)) + + # 创建登录折叠面板 + login_panels = [] + for site_name, records in login_site_data.items(): + # 按日期排序，最新的在前面 + try: + records.sort(key=lambda x: x.get("day_obj", datetime.now().date()), reverse=True) + except: + pass # 排序失败时跳过 + + # 获取最新的状态作为站点概要 + latest_status = records[0].get("status", "未知状态") + + # 确定状态颜色和图标 + status_color = "teal-lighten-3" + status_icon = "mdi-emoticon-happy-outline" + + if "失败" in latest_status or "错误" in latest_status: + status_color = "deep-orange-lighten-3" + status_icon = "mdi-emoticon-sad-outline" + elif "Cookie已失效" in latest_status: + status_color = "pink-lighten-3" + status_icon = "mdi-cookie-off" + elif "重试" in latest_status: + status_color = "amber-lighten-3" + status_icon = "mdi-emoticon-confused-outline" + elif "已签到" in latest_status: + status_color = "light-blue-lighten-3" + status_icon = "mdi-emoticon-cool-outline" + elif "成功" in latest_status: + status_color = "teal-lighten-3" + status_icon = "mdi-emoticon-happy-outline" + + # 创建每个站点的折叠面板 + login_panels.append( + self._create_expansion_panel(site_name, records, status_color, status_icon, latest_status)) + + # 添加样式 + return [ + { + 'component': 'style', + 'text': """ + .v-expansion-panel-title { + min-height: 48px !important; + padding: 0 16px !important; + } + .v-expansion-panel-text__wrapper { + padding: 0 !important; + } + .v-expansion-panel { + + margin-bottom: 10px !important; + box-shadow: 0 2px 8px rgba(0,0,0,0.05); + border-radius: 16px !important; + overflow: hidden !important; + border: 1px solid rgba(0,0,0,0.03); + transition: all 0.3s ease; + } + .v-expansion-panel:hover { + + box-shadow: 0 4px 12px rgba(0,0,0,0.08) !important; + transform: translateY(-2px); + } + .site-item { + border-radius: 10px; + transition: all 0.3s ease; + margin: 5px 0; + + } + .site-item:hover { + + transform: scale(1.01); + box-shadow: 0 2px 8px rgba(0,0,0,0.05); + } + .text-teal-lighten-3 { + color: #80CBC4 !important; + } + .text-deep-orange-lighten-3 { + color: #FFAB91 !important; + } + .text-pink-lighten-3 { + color: #F8BBD0 !important; + } + .text-amber-lighten-3 { + color: #FFE082 !important; + } + .text-light-blue-lighten-3 { + color: #81D4FA !important; + } + .status-icon { + width: 24px; + height: 24px; + line-height: 24px; + text-align: center; + border-radius: 50%; + margin-right: 8px; + } + .signin-card, .login-card { + transition: all 0.3s ease; + border-radius: 20px !important; + overflow: hidden; + box-shadow: 0 4px 15px rgba(0,0,0,0.03) !important; + border: 1px solid rgba(0,0,0,0.03); + } + .signin-card:hover, .login-card:hover { + transform: translateY(-3px); + box-shadow: 0 6px 20px rgba(0,0,0,0.05) !important; + } + .v-card-title.gradient-title { + margin-bottom: 0 !important; + border-bottom: 1px solid rgba(0,0,0,0.03); + } + .signin-card .v-card-title.gradient-title { + background: linear-gradient(135deg, rgba(128, 203, 196, 0.15) 0%, rgba(165, 214, 167, 0.15) 100%); + } + .login-card .v-card-title.gradient-title { + background: linear-gradient(135deg, rgba(129, 212, 250, 0.15) 0%, rgba(159, 168, 218, 0.15) 100%); + } + .date-chip { + margin: 2px !important; + border-radius: 14px !important; + font-size: 0.75rem !important; + } + .status-chip { + padding: 0 8px; + border-radius: 14px !important; + box-shadow: 0 2px 4px rgba(0,0,0,0.03); + } + .site-icon { + background: linear-gradient(45deg, #80CBC4, #81D4FA); + color: white !important; + border-radius: 12px; + width: 32px; + height: 32px; + display: flex; + align-items: center; + justify-content: center; + margin-right: 10px; + font-weight: bold; + font-size: 15px; + box-shadow: 0 2px 4px rgba(0,0,0,0.06); + } + .page-title { + font-size: 1.5rem; + font-weight: 600; + background: -webkit-linear-gradient(45deg, #80CBC4, #81D4FA); + -webkit-background-clip: text; + -webkit-text-fill-color: transparent; + } + """ + }, + { + 'component': 'VRow', + 'props': { + 'class': 'mt-2' + }, + 'content': [ + { + 'component': 'VCol', + 'props': { + 'cols': 12, + 'class': 'pb-0' + }, + 'content': [ + { + 'component': 'div', + 'props': { + 'class': 'd-flex align-center mb-4' + }, + 'content': [ + { + 'component': 'VIcon', + 'props': { + 'color': 'light-blue-lighten-3', + 'class': 'mr-2', + 'size': 'large', + 'icon': 'mdi-cat' + } + }, + { + 'component': 'h2', + 'props': { + 'class': 'page-title m-0' + }, + 'text': '站点签到小助手' + }, + { + 'component': 'VSpacer' + }, + { + 'component': 'VChip', + 'props': { + 'color': 'light-blue-lighten-5', + 'size': 'small', + 'variant': 'elevated', + 'class': 'ml-2', + 'prepend-icon': 'mdi-paw' + }, + 'text': f'显示 {len(sign_dates_list)} 天数据' + } + ] + } + ] + } + ] + }, + { + 'component': 'VRow', + 'content': [ + # 左侧 - 签到数据 + { + 'component': 'VCol', + 'props': { + 'cols': 12, + 'md': 6 + }, + 'content': [ + { + 'component': 'VCard', + 'props': { + 'variant': 'flat', + 'class': 'mb-4 signin-card' + }, + 'content': [ + { + 'component': 'VCardTitle', + 'props': { + 'class': 'gradient-title d-flex align-center pa-4' + }, + 'content': [ + { + 'component': 'VIcon', + 'props': { + 'class': 'mr-2', + 'color': 'teal-lighten-3', + 'size': 'small', + 'icon': 'mdi-duck' + } + }, + { + 'component': 'span', + 'props': { + 'class': 'font-weight-medium' + }, + 'text': '签到打卡记录' + }, + { + 'component': 'VSpacer' + }, + { + 'component': 'VChip', + 'props': { + 'color': 'teal-lighten-5', + 'size': 'x-small', + 'variant': 'elevated', + 'class': 'ml-2', + 'prepend-icon': 'mdi-rabbit' + }, + 'text': f'{len(signin_site_data)} 个站点' + } + ] + }, + { + 'component': 'VCardText', + 'props': { + 'class': 'pa-3' + }, + 'content': [ + { + 'component': 'VExpansionPanels', + 'props': { + 'variant': 'accordion', + 'class': 'mt-2' + }, + 'content': signin_panels or [{ + 'component': 'VAlert', + 'props': { + 'type': 'info', + 'text': '暂无签到数据', + 'variant': 'tonal', + 'class': 'mt-2', + 'density': 'compact', + 'prepend-icon': 'mdi-penguin' + } + }] + } + ] + } + ] + } + ] + }, + # 右侧 - 登录数据 + { + 'component': 'VCol', + 'props': { + 'cols': 12, + 'md': 6 + }, + 'content': [ + { + 'component': 'VCard', + 'props': { + 'variant': 'flat', + 'class': 'mb-4 login-card' + }, + 'content': [ + { + 'component': 'VCardTitle', + 'props': { + 'class': 'gradient-title d-flex align-center pa-4' + }, + 'content': [ + { + 'component': 'VIcon', + 'props': { + 'class': 'mr-2', + 'color': 'light-blue-accent-3', + 'size': 'small', + 'icon': 'mdi-dog' + } + }, + { + 'component': 'span', + 'props': { + 'class': 'font-weight-medium' + }, + 'text': '登录记录' + }, + { + 'component': 'VSpacer' + }, + { + 'component': 'VChip', + 'props': { + 'color': 'light-blue-lighten-4', + 'size': 'x-small', + 'variant': 'elevated', + 'class': 'ml-2', + 'prepend-icon': 'mdi-panda' + }, + 'text': f'{len(login_site_data)} 个站点' + } + ] + }, + { + 'component': 'VCardText', + 'props': { + 'class': 'pa-3' + }, + 'content': [ + { + 'component': 'VExpansionPanels', + 'props': { + 'variant': 'accordion', + 'class': 'mt-2' + }, + 'content': login_panels or [{ + 'component': 'VAlert', + 'props': { + 'type': 'info', + 'text': '暂无登录数据', + 'variant': 'tonal', + 'class': 'mt-2', + 'density': 'compact', + 'prepend-icon': 'mdi-cat' + } + }] + } + ] + } + ] + } + ] + } + ] + } + ] + + def _create_expansion_panel(self, site_name, records, status_color, status_icon, latest_status): + """创建站点折叠面板""" + # 生成站点图标（使用站点名的首字母） + site_initial = site_name[0].upper() if site_name else "?" + + # 生成记录列表 + records_list = [] + for record in records: + date_str = record.get("date", "") + status_text = record.get("status", "未知状态") + + # 确定状态颜色和图标 + record_color = "success" + record_icon = "mdi-check-circle" + + if "失败" in status_text or "错误" in status_text: + record_color = "error" + record_icon = "mdi-alert-circle" + elif "Cookie已失效" in status_text: + record_color = "error" + record_icon = "mdi-cookie-off" + elif "重试" in status_text: + record_color = "warning" + record_icon = "mdi-refresh" + elif "已签到" in status_text: + record_color = "info" + record_icon = "mdi-check" + elif "登录成功" in status_text: + record_color = "success" + record_icon = "mdi-login-variant" + + # 创建记录项 + records_list.append({ + 'component': 'VListItem', + 'props': { + 'class': 'site-item px-2 py-1' + }, + 'content': [ + { + 'component': 'div', + 'props': { + 'class': 'd-flex align-center w-100' + }, + 'content': [ + { + 'component': 'VChip', + 'props': { + 'color': 'grey-lighten-3', + 'size': 'x-small', + 'class': 'date-chip mr-2', + 'variant': 'flat', + 'prepend-icon': 'mdi-flower-tulip' + }, + 'text': date_str + }, + { + 'component': 'VSpacer' + }, + { + 'component': 'VChip', + 'props': { + 'color': record_color, + 'size': 'x-small', + 'class': 'ml-2 status-chip', + 'variant': 'flat', + 'prepend-icon': record_icon + }, + 'text': status_text + } + ] + } + ] + }) + + # 创建折叠面板 + return { + 'component': 'VExpansionPanel', + 'content': [ + { + 'component': 'VExpansionPanelTitle', + 'content': [{ + 'component': 'div', + 'props': { + 'class': 'd-flex align-center' + }, + 'content': [ + { + 'component': 'div', + 'props': { + 'class': 'site-icon' + }, + 'text': site_initial + }, + { + 'component': 'span', + 'props': { + 'class': 'font-weight-medium' + }, + 'text': site_name + }, + { + 'component': 'VSpacer' + }, + { + 'component': 'VIcon', + 'props': { + 'color': status_color, + 'class': 'mr-2', + 'size': 'small' + }, + 'text': status_icon + }, + { + 'component': 'span', + 'props': { + 'class': f'text-{status_color} text-caption' + }, + 'text': latest_status + } + ] + }] + }, + { + 'component': 'VExpansionPanelText', + 'content': [ + { + 'component': 'VList', + 'props': { + 'lines': 'one', + 'density': 'compact' + }, + 'content': records_list + } + ] + } + ] + } + + @eventmanager.register(EventType.PluginAction) + def sign_in(self, event: Event = None): + """ + 自动签到|模拟登录 + """ + if event: + event_data = event.event_data + if not event_data or event_data.get("action") != "site_signin": + return + # 日期 + today = datetime.today() + if self._start_time and self._end_time: + if int(datetime.today().hour) < self._start_time or int(datetime.today().hour) > self._end_time: + logger.error( + f"当前时间 {int(datetime.today().hour)} 不在 {self._start_time}-{self._end_time} 范围内，暂不执行任务") + return + if event: + logger.info("收到命令，开始站点签到 ...") + self.post_message(channel=event.event_data.get("channel"), + title="开始站点签到 ...", + userid=event.event_data.get("user")) + + if self._sign_sites: + self.__do(today=today, type_str="签到", do_sites=self._sign_sites, event=event) + if self._login_sites: + self.__do(today=today, type_str="登录", do_sites=self._login_sites, event=event) + + def __do(self, today: datetime, type_str: str, do_sites: list, event: Event = None): + """ + 签到逻辑 + """ + last_day = today - timedelta(days=4) + last_day_str = last_day.strftime('%Y-%m-%d') + # 删除昨天历史 + self.del_data(key=type_str + "-" + last_day_str) + self.del_data(key=f"{last_day.month}月{last_day.day}日") + + # 查看今天有没有签到|登录历史 + today = today.strftime('%Y-%m-%d') + today_history = self.get_data(key=type_str + "-" + today) + + # 查询所有站点 + all_sites = [site for site in self.sites.get_indexers() if not site.get("public")] + self.__custom_sites() + # 过滤掉没有选中的站点 + if do_sites: + do_sites = [site for site in all_sites if site.get("id") in do_sites] + else: + do_sites = all_sites + + # 今日没数据 + if not today_history or self._clean: + logger.info(f"今日 {today} 未{type_str}，开始{type_str}已选站点") + if self._clean: + # 关闭开关 + self._clean = False + else: + # 需要重试站点 + retry_sites = today_history.get("retry") or [] + # 今天已签到|登录站点 + already_sites = today_history.get("do") or [] + + # 今日未签|登录站点 + no_sites = [site for site in do_sites if + site.get("id") not in already_sites or site.get("id") in retry_sites] + + if not no_sites: + logger.info(f"今日 {today} 已{type_str}，无重新{type_str}站点，本次任务结束") + return + + # 任务站点 = 需要重试+今日未do + do_sites = no_sites + logger.info(f"今日 {today} 已{type_str}，开始重试命中关键词站点") + + if not do_sites: + logger.info(f"没有需要{type_str}的站点") + return + + # 执行签到 + logger.info(f"开始执行{type_str}任务 ...") + if type_str == "签到": + with ThreadPool(min(len(do_sites), int(self._queue_cnt))) as p: + status = p.map(self.signin_site, do_sites) + else: + with ThreadPool(min(len(do_sites), int(self._queue_cnt))) as p: + status = p.map(self.login_site, do_sites) + + if status: + logger.info(f"站点{type_str}任务完成！") + # 获取今天的日期 + key = f"{datetime.now().month}月{datetime.now().day}日" + today_data = self.get_data(key) + if today_data: + if not isinstance(today_data, list): + today_data = [today_data] + for s in status: + today_data.append({ + "site": s[0], + "status": s[1] + }) + else: + today_data = [{ + "site": s[0], + "status": s[1] + } for s in status] + # 保存数据 + self.save_data(key, today_data) + + # 命中重试词的站点id + retry_sites = [] + # 命中重试词的站点签到msg + retry_msg = [] + # 登录成功 + login_success_msg = [] + # 签到成功 + sign_success_msg = [] + # 已签到 + already_sign_msg = [] + # 仿真签到成功 + fz_sign_msg = [] + # 失败｜错误 + failed_msg = [] + + sites = {site.get('name'): site.get("id") for site in self.sites.get_indexers() if not site.get("public")} + for s in status: + site_name = s[0] + site_id = None + if site_name: + site_id = sites.get(site_name) + + if 'Cookie已失效' in str(s) and site_id: + # 触发自动登录插件登录 + logger.info(f"触发站点 {site_name} 自动登录更新Cookie和Ua") + self.eventmanager.send_event(EventType.PluginAction, + { + "site_id": site_id, + "action": "site_refresh" + }) + # 记录本次命中重试关键词的站点 + if self._retry_keyword: + if site_id: + match = re.search(self._retry_keyword, s[1]) + if match: + logger.debug(f"站点 {site_name} 命中重试关键词 {self._retry_keyword}") + retry_sites.append(site_id) + # 命中的站点 + retry_msg.append(s) + continue + + if "登录成功" in str(s): + login_success_msg.append(s) + elif "仿真签到成功" in str(s): + fz_sign_msg.append(s) + continue + elif "签到成功" in str(s): + sign_success_msg.append(s) + elif '已签到' in str(s): + already_sign_msg.append(s) + else: + failed_msg.append(s) + + if not self._retry_keyword: + # 没设置重试关键词则重试已选站点 + retry_sites = self._sign_sites if type_str == "签到" else self._login_sites + logger.debug(f"下次{type_str}重试站点 {retry_sites}") + + # 存入历史 + self.save_data(key=type_str + "-" + today, + value={ + "do": self._sign_sites if type_str == "签到" else self._login_sites, + "retry": retry_sites + }) + + # 自动Cloudflare IP优选 + if self._auto_cf and int(self._auto_cf) > 0 and retry_msg and len(retry_msg) >= int(self._auto_cf): + self.eventmanager.send_event(EventType.PluginAction, { + "action": "cloudflare_speedtest" + }) + + # 发送通知 + if self._notify: + # 签到详细信息 登录成功、签到成功、已签到、仿真签到成功、失败--命中重试 + signin_message = login_success_msg + sign_success_msg + already_sign_msg + fz_sign_msg + failed_msg + if len(retry_msg) > 0: + signin_message += retry_msg + + signin_message = "\n".join([f'【{s[0]}】{s[1]}' for s in signin_message if s]) + self.post_message(title=f"【站点自动{type_str}】", + mtype=NotificationType.SiteMessage, + text=f"全部{type_str}数量: {len(self._sign_sites if type_str == '签到' else self._login_sites)} \n" + f"本次{type_str}数量: {len(do_sites)} \n" + f"下次{type_str}数量: {len(retry_sites) if self._retry_keyword else 0} \n" + f"{signin_message}" + ) + if event: + self.post_message(channel=event.event_data.get("channel"), + title=f"站点{type_str}完成！", userid=event.event_data.get("user")) + else: + logger.error(f"站点{type_str}任务失败！") + if event: + self.post_message(channel=event.event_data.get("channel"), + title=f"站点{type_str}任务失败！", userid=event.event_data.get("user")) + # 保存配置 + self.__update_config() + + def __build_class(self, url) -> Any: + for site_schema in self._site_schema: + try: + if site_schema.match(url): + return site_schema + except Exception as e: + logger.error("站点模块加载失败：%s" % str(e)) + return None + + def signin_by_domain(self, url: str, apikey: str) -> schemas.Response: + """ + 签到一个站点，可由API调用 + """ + # 校验 + if apikey != settings.API_TOKEN: + return schemas.Response(success=False, message="API密钥错误") + domain = StringUtils.get_url_domain(url) + site_info = self.sites.get_indexer(domain) + if not site_info: + return schemas.Response( + success=True, + message=f"站点【{url}】不存在" + ) + else: + site_name, message = self.signin_site(site_info) + return schemas.Response( + success=True, + message=f"站点【{site_name}】{message or '签到成功'}" + ) + + def signin_site(self, site_info: CommentedMap) -> Tuple[str, str]: + """ + 签到一个站点 + """ + site_module = self.__build_class(site_info.get("url")) + # 开始记时 + start_time = datetime.now() + if site_module and hasattr(site_module, "signin"): + try: + state, message = site_module().signin(site_info) + except Exception as e: + traceback.print_exc() + state, message = False, f"签到失败：{str(e)}" + else: + state, message = self.__signin_base(site_info) + # 统计 + seconds = (datetime.now() - start_time).seconds + domain = StringUtils.get_url_domain(site_info.get('url')) + if state: + self.siteoper.success(domain=domain, seconds=seconds) + else: + self.siteoper.fail(domain) + return site_info.get("name"), message + + @staticmethod + def __signin_base(site_info: CommentedMap) -> Tuple[bool, str]: + """ + 通用签到处理 + :param site_info: 站点信息 + :return: 签到结果信息 + """ + if not site_info: + return False, "" + site = site_info.get("name") + site_url = site_info.get("url") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + render = site_info.get("render") + proxies = settings.PROXY if site_info.get("proxy") else None + proxy_server = settings.PROXY_SERVER if site_info.get("proxy") else None + if not site_url or not site_cookie: + logger.warn(f"未配置 {site} 的站点地址或Cookie，无法签到") + return False, "" + # 模拟登录 + try: + # 访问链接 + checkin_url = site_url + if site_url.find("attendance.php") == -1: + # 拼登签到地址 + checkin_url = urljoin(site_url, "attendance.php") + logger.info(f"开始站点签到：{site}，地址：{checkin_url}...") + if render: + page_source = PlaywrightHelper().get_page_source(url=checkin_url, + cookies=site_cookie, + ua=ua, + proxies=proxy_server) + if not SiteUtils.is_logged_in(page_source): + if under_challenge(page_source): + return False, f"无法通过Cloudflare！" + return False, f"仿真登录失败，Cookie已失效！" + else: + # 判断是否已签到 + if re.search(r'已签|签到已得', page_source, re.IGNORECASE) \ + or SiteUtils.is_checkin(page_source): + return True, f"签到成功" + return True, "仿真签到成功" + else: + res = RequestUtils(cookies=site_cookie, + ua=ua, + proxies=proxies + ).get_res(url=checkin_url) + if not res and site_url != checkin_url: + logger.info(f"开始站点模拟登录：{site}，地址：{site_url}...") + res = RequestUtils(cookies=site_cookie, + ua=ua, + proxies=proxies + ).get_res(url=site_url) + # 判断登录状态 + if res and res.status_code in [200, 500, 403]: + if not SiteUtils.is_logged_in(res.text): + if under_challenge(res.text): + msg = "站点被Cloudflare防护，请打开站点浏览器仿真" + elif res.status_code == 200: + msg = "Cookie已失效" + else: + msg = f"状态码：{res.status_code}" + logger.warn(f"{site} 签到失败，{msg}") + return False, f"签到失败，{msg}！" + else: + logger.info(f"{site} 签到成功") + return True, f"签到成功" + elif res is not None: + logger.warn(f"{site} 签到失败，状态码：{res.status_code}") + return False, f"签到失败，状态码：{res.status_code}！" + else: + logger.warn(f"{site} 签到失败，无法打开网站") + return False, f"签到失败，无法打开网站！" + except Exception as e: + logger.warn("%s 签到失败：%s" % (site, str(e))) + traceback.print_exc() + return False, f"签到失败：{str(e)}！" + + def login_site(self, site_info: CommentedMap) -> Tuple[str, str]: + """ + 模拟登录一个站点 + """ + site_module = self.__build_class(site_info.get("url")) + # 开始记时 + start_time = datetime.now() + if site_module and hasattr(site_module, "login"): + try: + state, message = site_module().login(site_info) + except Exception as e: + traceback.print_exc() + state, message = False, f"模拟登录失败：{str(e)}" + else: + state, message = self.__login_base(site_info) + # 统计 + seconds = (datetime.now() - start_time).seconds + domain = StringUtils.get_url_domain(site_info.get('url')) + if state: + self.siteoper.success(domain=domain, seconds=seconds) + else: + self.siteoper.fail(domain) + return site_info.get("name"), message + + @staticmethod + def __login_base(site_info: CommentedMap) -> Tuple[bool, str]: + """ + 模拟登录通用处理 + :param site_info: 站点信息 + :return: 签到结果信息 + """ + if not site_info: + return False, "" + site = site_info.get("name") + site_url = site_info.get("url") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + render = site_info.get("render") + proxies = settings.PROXY if site_info.get("proxy") else None + proxy_server = settings.PROXY_SERVER if site_info.get("proxy") else None + if not site_url or not site_cookie: + logger.warn(f"未配置 {site} 的站点地址或Cookie，无法签到") + return False, "" + # 模拟登录 + try: + # 访问链接 + site_url = str(site_url).replace("attendance.php", "") + logger.info(f"开始站点模拟登录：{site}，地址：{site_url}...") + if render: + page_source = PlaywrightHelper().get_page_source(url=site_url, + cookies=site_cookie, + ua=ua, + proxies=proxy_server) + if not SiteUtils.is_logged_in(page_source): + if under_challenge(page_source): + return False, f"无法通过Cloudflare！" + return False, f"仿真登录失败，Cookie已失效！" + else: + return True, "模拟登录成功" + else: + res = RequestUtils(cookies=site_cookie, + ua=ua, + proxies=proxies + ).get_res(url=site_url) + # 判断登录状态 + if res and res.status_code in [200, 500, 403]: + if not SiteUtils.is_logged_in(res.text): + if under_challenge(res.text): + msg = "站点被Cloudflare防护，请打开站点浏览器仿真" + elif res.status_code == 200: + msg = "Cookie已失效" + else: + msg = f"状态码：{res.status_code}" + logger.warn(f"{site} 模拟登录失败，{msg}") + return False, f"模拟登录失败，{msg}！" + else: + logger.info(f"{site} 模拟登录成功") + return True, f"模拟登录成功" + elif res is not None: + logger.warn(f"{site} 模拟登录失败，状态码：{res.status_code}") + return False, f"模拟登录失败，状态码：{res.status_code}！" + else: + logger.warn(f"{site} 模拟登录失败，无法打开网站") + return False, f"模拟登录失败，无法打开网站！" + except Exception as e: + logger.warn("%s 模拟登录失败：%s" % (site, str(e))) + traceback.print_exc() + return False, f"模拟登录失败：{str(e)}！" + + def stop_service(self): + """ + 退出插件 + """ + try: + if self._scheduler: + self._scheduler.remove_all_jobs() + if self._scheduler.running: + self._scheduler.shutdown() + self._scheduler = None + except Exception as e: + logger.error("退出插件失败：%s" % str(e)) + + @eventmanager.register(EventType.SiteDeleted) + def site_deleted(self, event): + """ + 删除对应站点选中 + """ + site_id = event.event_data.get("site_id") + config = self.get_config() + if config: + self._sign_sites = self.__remove_site_id(config.get("sign_sites") or [], site_id) + self._login_sites = self.__remove_site_id(config.get("login_sites") or [], site_id) + # 保存配置 + self.__update_config() + + def __remove_site_id(self, do_sites, site_id): + if do_sites: + if isinstance(do_sites, str): + do_sites = [do_sites] + + # 删除对应站点 + if site_id: + do_sites = [site for site in do_sites if int(site) != int(site_id)] + else: + # 清空 + do_sites = [] + + # 若无站点，则停止 + if len(do_sites) == 0: + self._enabled = False + + return do_sites + + +def record_to_row(record): + """辅助函数：将记录转换为表格行""" + status = record.get("status", "") + + # 确定状态图标和颜色 + icon = "mdi-check-circle" + color = "success" + + if "失败" in status or "错误" in status: + icon = "mdi-alert-circle" + color = "error" + elif "Cookie已失效" in status: + icon = "mdi-cookie-off" + color = "error" + elif "已签到" in status: + icon = "mdi-check" + color = "grey" + elif "成功" in status: + icon = "mdi-check-circle" + color = "success" + + return { + 'component': 'tr', + 'props': { + 'class': 'text-sm' + }, + 'content': [ + { + 'component': 'td', + 'props': { + 'class': 'text-start' + }, + 'text': record.get("date", "") + }, + { + 'component': 'td', + 'props': { + 'class': 'text-start' + }, + 'text': status + }, + { + 'component': 'td', + 'props': { + 'class': 'text-center' + }, + 'content': [ + { + 'component': 'VIcon', + 'props': { + 'color': color, + 'size': 'small' + }, + 'text': icon + } + ] + } + ] + } diff --git a/plugins.v2/autosignin/sites/52pt.py b/plugins.v2/autosignin/sites/52pt.py new file mode 100644 index 0000000..44c6155 --- /dev/null +++ b/plugins.v2/autosignin/sites/52pt.py @@ -0,0 +1,147 @@ +import random +import re +from typing import Tuple + +from lxml import etree + +from app.core.config import settings +from app.log import logger +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.http import RequestUtils +from app.utils.string import StringUtils + + +class Pt52(_ISiteSigninHandler): + """ + 52pt + 如果填写openai key则调用chatgpt获取答案 + 否则随机 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "52pt.site" + + # 已签到 + _sign_regex = ['今天已经签过到了'] + + # 签到成功，待补充 + _success_regex = ['\\d+点魔力值'] + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if StringUtils.url_equal(url, cls.site_url) else False + + def signin(self, site_info: dict) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + site = site_info.get("name") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + render = site_info.get("render") + proxy = site_info.get("proxy") + + # 判断今日是否已签到 + html_text = self.get_page_source(url='https://52pt.site/bakatest.php', + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + + if not html_text: + logger.error(f"{site} 签到失败，请检查站点连通性") + return False, '签到失败，请检查站点连通性' + + if "login.php" in html_text: + logger.error(f"{site} 签到失败，Cookie已失效") + return False, '签到失败，Cookie已失效' + + sign_status = self.sign_in_result(html_res=html_text, + regexs=self._sign_regex) + if sign_status: + logger.info(f"今日已签到") + return True, '今日已签到' + + # 没有签到则解析html + html = etree.HTML(html_text) + + if not html: + return False, '签到失败' + + # 获取页面问题、答案 + questionid = html.xpath("//input[@name='questionid']/@value")[0] + option_ids = html.xpath("//input[@name='choice[]']/@value") + question_str = html.xpath("//td[@class='text' and contains(text(),'请问：')]/text()")[0] + + # 正则获取问题 + match = re.search(r'请问：(.+)', question_str) + if match: + question_str = match.group(1) + logger.debug(f"获取到签到问题 {question_str}") + else: + logger.error(f"未获取到签到问题") + return False, f"【{site}】签到失败，未获取到签到问题" + + # 正确答案，默认随机，如果gpt返回则用gpt返回的答案提交 + choice = [option_ids[random.randint(0, len(option_ids) - 1)]] + + # 签到 + return self.__signin(questionid=questionid, + choice=choice, + site_cookie=site_cookie, + ua=ua, + proxy=proxy, + site=site) + + def __signin(self, questionid: str, + choice: list, + site: str, + site_cookie: str, + ua: str, + proxy: bool) -> Tuple[bool, str]: + """ + 签到请求 + questionid: 450 + choice[]: 8 + choice[]: 4 + usercomment: 此刻心情:无 + submit: 提交 + 多选会有多个choice[].... + """ + data = { + 'questionid': questionid, + 'choice[]': choice[0] if len(choice) == 1 else choice, + 'usercomment': '太难了！', + 'wantskip': '不会' + } + logger.debug(f"签到请求参数 {data}") + + sign_res = RequestUtils(cookies=site_cookie, + ua=ua, + proxies=settings.PROXY if proxy else None + ).post_res(url='https://52pt.site/bakatest.php', data=data) + if not sign_res or sign_res.status_code != 200: + logger.error(f"{site} 签到失败，签到接口请求失败") + return False, '签到失败，签到接口请求失败' + + # 判断是否签到成功 + sign_status = self.sign_in_result(html_res=sign_res.text, + regexs=self._success_regex) + if sign_status: + logger.info(f"{site} 签到成功") + return True, '签到成功' + else: + sign_status = self.sign_in_result(html_res=sign_res.text, + regexs=self._sign_regex) + if sign_status: + logger.info(f"{site} 今日已签到") + return True, '今日已签到' + + logger.error(f"{site} 签到失败，请到页面查看") + return False, '签到失败，请到页面查看' diff --git a/plugins.v2/autosignin/sites/__init__.py b/plugins.v2/autosignin/sites/__init__.py new file mode 100644 index 0000000..b0e2ef2 --- /dev/null +++ b/plugins.v2/autosignin/sites/__init__.py @@ -0,0 +1,99 @@ +# -*- coding: utf-8 -*- +import re +from abc import ABCMeta, abstractmethod +from typing import Tuple + +import chardet +from ruamel.yaml import CommentedMap + +from app.core.config import settings +from app.helper.browser import PlaywrightHelper +from app.log import logger +from app.utils.http import RequestUtils +from app.utils.string import StringUtils + + +class _ISiteSigninHandler(metaclass=ABCMeta): + """ + 实现站点签到的基类，所有站点签到类都需要继承此类，并实现match和signin方法 + 实现类放置到sitesignin目录下将会自动加载 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "" + + @abstractmethod + def match(self, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + if StringUtils.url_equal(url, self.site_url): + return True + return False + + @abstractmethod + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: True|False,签到结果信息 + """ + pass + + @staticmethod + def get_page_source(url: str, cookie: str, ua: str, proxy: bool, render: bool, token: str = None) -> str: + """ + 获取页面源码 + :param url: Url地址 + :param cookie: Cookie + :param ua: UA + :param proxy: 是否使用代理 + :param render: 是否渲染 + :param token: JWT Token + :return: 页面源码，错误信息 + """ + if render: + return PlaywrightHelper().get_page_source(url=url, + cookies=cookie, + ua=ua, + proxies=settings.PROXY_SERVER if proxy else None) + else: + if token: + headers = { + "Authorization": token, + "User-Agent": ua + } + else: + headers = { + "User-Agent": ua, + "Cookie": cookie + } + res = RequestUtils(headers=headers, + proxies=settings.PROXY if proxy else None).get_res(url=url) + if res is not None: + # 使用chardet检测字符编码 + raw_data = res.content + if raw_data: + try: + result = chardet.detect(raw_data) + encoding = result['encoding'] + # 解码为字符串 + return raw_data.decode(encoding) + except Exception as e: + logger.error(f"chardet解码失败：{str(e)}") + return res.text + else: + return res.text + return "" + + @staticmethod + def sign_in_result(html_res: str, regexs: list) -> bool: + """ + 判断是否签到成功 + """ + html_text = re.sub(r"#\d+", "", re.sub(r"\d+px", "", html_res)) + for regex in regexs: + if re.search(str(regex), html_text): + return True + return False diff --git a/plugins.v2/autosignin/sites/btschool.py b/plugins.v2/autosignin/sites/btschool.py new file mode 100644 index 0000000..b8f2671 --- /dev/null +++ b/plugins.v2/autosignin/sites/btschool.py @@ -0,0 +1,75 @@ +from typing import Tuple + +from ruamel.yaml import CommentedMap + +from app.log import logger +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.string import StringUtils + + +class BTSchool(_ISiteSigninHandler): + """ + 学校签到 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "pt.btschool.club" + + # 已签到 + _sign_text = '每日签到' + + @classmethod + def match(cls, url) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if StringUtils.url_equal(url, cls.site_url) else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + site = site_info.get("name") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + render = site_info.get("render") + proxy = site_info.get("proxy") + + logger.info(f"{site} 开始签到") + # 判断今日是否已签到 + html_text = self.get_page_source(url='https://pt.btschool.club', + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + + if not html_text: + logger.error(f"{site} 签到失败，请检查站点连通性") + return False, '签到失败，请检查站点连通性' + + if "login.php" in html_text: + logger.error(f"{site} 签到失败，Cookie已失效") + return False, '签到失败，Cookie已失效' + + # 已签到 + if self._sign_text not in html_text: + logger.info(f"{site} 今日已签到") + return True, '今日已签到' + + html_text = self.get_page_source(url='https://pt.btschool.club/index.php?action=addbonus', + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + + if not html_text: + logger.error(f"{site} 签到失败，签到接口请求失败") + return False, '签到失败，签到接口请求失败' + + # 签到成功 + if self._sign_text not in html_text: + logger.info(f"{site} 签到成功") + return True, '签到成功' diff --git a/plugins.v2/autosignin/sites/chdbits.py b/plugins.v2/autosignin/sites/chdbits.py new file mode 100644 index 0000000..ed2cf67 --- /dev/null +++ b/plugins.v2/autosignin/sites/chdbits.py @@ -0,0 +1,148 @@ +import random +import re +from typing import Tuple + +from lxml import etree +from ruamel.yaml import CommentedMap + +from app.core.config import settings +from app.log import logger +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.http import RequestUtils +from app.utils.string import StringUtils + + +class CHDBits(_ISiteSigninHandler): + """ + 彩虹岛签到 + 如果填写openai key则调用chatgpt获取答案 + 否则随机 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "ptchdbits.co" + + # 已签到 + _sign_regex = ['今天已经签过到了'] + + # 签到成功，待补充 + _success_regex = ['\\d+点魔力值'] + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if StringUtils.url_equal(url, cls.site_url) else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + site = site_info.get("name") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + proxy = site_info.get("proxy") + render = site_info.get("render") + + # 判断今日是否已签到 + html_text = self.get_page_source(url='https://ptchdbits.co/bakatest.php', + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + + if not html_text: + logger.error(f"{site} 签到失败，请检查站点连通性") + return False, '签到失败，请检查站点连通性' + + if "login.php" in html_text: + logger.error(f"{site} 签到失败，Cookie已失效") + return False, '签到失败，Cookie已失效' + + sign_status = self.sign_in_result(html_res=html_text, + regexs=self._sign_regex) + if sign_status: + logger.info(f"{site} 今日已签到") + return True, '今日已签到' + + # 没有签到则解析html + html = etree.HTML(html_text) + + if not html: + return False, '签到失败' + + # 获取页面问题、答案 + questionid = html.xpath("//input[@name='questionid']/@value")[0] + option_ids = html.xpath("//input[@name='choice[]']/@value") + question_str = html.xpath("//td[@class='text' and contains(text(),'请问：')]/text()")[0] + + # 正则获取问题 + match = re.search(r'请问：(.+)', question_str) + if match: + question_str = match.group(1) + logger.debug(f"获取到签到问题 {question_str}") + else: + logger.error(f"未获取到签到问题") + return False, f"【{site}】签到失败，未获取到签到问题" + + # 正确答案，默认随机，如果gpt返回则用gpt返回的答案提交 + choice = [option_ids[random.randint(0, len(option_ids) - 1)]] + + # 签到 + return self.__signin(questionid=questionid, + choice=choice, + site_cookie=site_cookie, + ua=ua, + proxy=proxy, + site=site) + + def __signin(self, questionid: str, + choice: list, + site: str, + site_cookie: str, + ua: str, + proxy: bool) -> Tuple[bool, str]: + """ + 签到请求 + questionid: 450 + choice[]: 8 + choice[]: 4 + usercomment: 此刻心情:无 + submit: 提交 + 多选会有多个choice[].... + """ + data = { + 'questionid': questionid, + 'choice[]': choice[0] if len(choice) == 1 else choice, + 'usercomment': '太难了！', + 'wantskip': '不会' + } + logger.debug(f"签到请求参数 {data}") + + sign_res = RequestUtils(cookies=site_cookie, + ua=ua, + proxies=settings.PROXY if proxy else None + ).post_res(url='https://ptchdbits.co/bakatest.php', data=data) + if not sign_res or sign_res.status_code != 200: + logger.error(f"{site} 签到失败，签到接口请求失败") + return False, '签到失败，签到接口请求失败' + + # 判断是否签到成功 + sign_status = self.sign_in_result(html_res=sign_res.text, + regexs=self._success_regex) + if sign_status: + logger.info(f"{site} 签到成功") + return True, '签到成功' + else: + sign_status = self.sign_in_result(html_res=sign_res.text, + regexs=self._sign_regex) + if sign_status: + logger.info(f"{site} 今日已签到") + return True, '今日已签到' + + logger.error(f"{site} 签到失败，请到页面查看") + return False, '签到失败，请到页面查看' diff --git a/plugins.v2/autosignin/sites/haidan.py b/plugins.v2/autosignin/sites/haidan.py new file mode 100644 index 0000000..23f6b03 --- /dev/null +++ b/plugins.v2/autosignin/sites/haidan.py @@ -0,0 +1,70 @@ +from typing import Tuple + +from ruamel.yaml import CommentedMap + +from app.log import logger +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.string import StringUtils + + +class HaiDan(_ISiteSigninHandler): + """ + 海胆签到 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "haidan.video" + + # 签到成功 + _succeed_regex = ['(?<=value=")已经打卡(?=")'] + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if StringUtils.url_equal(url, cls.site_url) else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + site = site_info.get("name") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + proxy = site_info.get("proxy") + render = site_info.get("render") + + # 签到 + # 签到页会重定向到index.php，由于302重定向特性，导致index.php没有携带cookie + self.get_page_source(url='https://www.haidan.video/signin.php', + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + + # 重新携带cookie获取index.php查看签到结果 + html_text = self.get_page_source(url='https://www.haidan.video/index.php', + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + if not html_text: + logger.error(f"{site} 签到失败，请检查站点连通性") + return False, '签到失败，请检查站点连通性' + + if "login.php" in html_text: + logger.error(f"{site} 签到失败，Cookie已失效") + return False, '签到失败，Cookie已失效' + + sign_status = self.sign_in_result(html_res=html_text, + regexs=self._succeed_regex) + if sign_status: + logger.info(f"{site} 签到成功") + return True, '签到成功' + + logger.error(f"{site} 签到失败，签到接口返回 {html_text}") + return False, '签到失败' diff --git a/plugins.v2/autosignin/sites/hares.py b/plugins.v2/autosignin/sites/hares.py new file mode 100644 index 0000000..5aea8f1 --- /dev/null +++ b/plugins.v2/autosignin/sites/hares.py @@ -0,0 +1,83 @@ +import json +from typing import Tuple + +from ruamel.yaml import CommentedMap + +from app.core.config import settings +from app.log import logger +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.http import RequestUtils +from app.utils.string import StringUtils + + +class Hares(_ISiteSigninHandler): + """ + 白兔签到 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "club.hares.top" + + # 已签到 + _sign_text = '已签到' + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if StringUtils.url_equal(url, cls.site_url) else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + site = site_info.get("name") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + proxy = site_info.get("proxy") + render = site_info.get("render") + + # 获取页面html + html_text = self.get_page_source(url='https://club.hares.top', + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + + if not html_text: + logger.error(f"{site} 模拟访问失败，请检查站点连通性") + return False, '模拟访问失败，请检查站点连通性' + + if "login.php" in html_text: + logger.error(f"{site} 模拟访问失败，Cookie已失效") + return False, '模拟访问失败，Cookie已失效' + + # if self._sign_text in html_res.text: + # logger.info(f"今日已签到") + # return True, '今日已签到' + + headers = { + 'Accept': 'application/json', + "User-Agent": ua + } + sign_res = RequestUtils(cookies=site_cookie, + headers=headers, + proxies=settings.PROXY if proxy else None + ).get_res(url="https://club.hares.top/attendance.php?action=sign") + if not sign_res or sign_res.status_code != 200: + logger.error(f"{site} 签到失败，签到接口请求失败") + return False, '签到失败，签到接口请求失败' + + # {"code":1,"msg":"您今天已经签到过了"} + # {"code":0,"msg":"签到成功"} + sign_dict = json.loads(sign_res.text) + if sign_dict['code'] == 0: + logger.info(f"{site} 签到成功") + return True, '签到成功' + else: + logger.info(f"{site} 今日已签到") + return True, '今日已签到' diff --git a/plugins.v2/autosignin/sites/hdarea.py b/plugins.v2/autosignin/sites/hdarea.py new file mode 100644 index 0000000..d88800a --- /dev/null +++ b/plugins.v2/autosignin/sites/hdarea.py @@ -0,0 +1,69 @@ +from typing import Tuple + +from ruamel.yaml import CommentedMap + +from app.core.config import settings +from app.log import logger +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.http import RequestUtils +from app.utils.string import StringUtils + + +class HDArea(_ISiteSigninHandler): + """ + 好大签到 + """ + + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "hdarea.club" + + # 签到成功 + _success_text = "此次签到您获得" + _repeat_text = "请不要重复签到哦" + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if StringUtils.url_equal(url, cls.site_url) else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + site = site_info.get("name") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + proxies = settings.PROXY if site_info.get("proxy") else None + + # 获取页面html + data = { + 'action': 'sign_in' + } + html_res = RequestUtils(cookies=site_cookie, + ua=ua, + proxies=proxies + ).post_res(url="https://hdarea.club/sign_in.php", data=data) + if not html_res or html_res.status_code != 200: + logger.error(f"{site} 签到失败，请检查站点连通性") + return False, '签到失败，请检查站点连通性' + + if "login.php" in html_res.text: + logger.error(f"{site} 签到失败，Cookie已失效") + return False, '签到失败，Cookie已失效' + + # 判断是否已签到 + # '已连续签到278天，此次签到您获得了100魔力值奖励!' + if self._success_text in html_res.text: + logger.info(f"{site} 签到成功") + return True, '签到成功' + if self._repeat_text in html_res.text: + logger.info(f"{site} 今日已签到") + return True, '今日已签到' + logger.error(f"{site} 签到失败，签到接口返回 {html_res.text}") + return False, '签到失败' diff --git a/plugins.v2/autosignin/sites/hdchina.py b/plugins.v2/autosignin/sites/hdchina.py new file mode 100644 index 0000000..1d14982 --- /dev/null +++ b/plugins.v2/autosignin/sites/hdchina.py @@ -0,0 +1,117 @@ +import json +from typing import Tuple + +from lxml import etree +from ruamel.yaml import CommentedMap + +from app.core.config import settings +from app.log import logger +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.http import RequestUtils +from app.utils.string import StringUtils + + +class HDChina(_ISiteSigninHandler): + """ + 瓷器签到 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "hdchina.org" + + # 已签到 + _sign_regex = ['已签到'] + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if StringUtils.url_equal(url, cls.site_url) else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + site = site_info.get("name") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + proxies = settings.PROXY if site_info.get("proxy") else None + + # 尝试解决瓷器cookie每天签到后过期,只保留hdchina=部分 + cookie = "" + # 按照分号进行字符串拆分 + sub_strs = site_cookie.split(";") + # 遍历每个子字符串 + for sub_str in sub_strs: + if "hdchina=" in sub_str: + # 如果子字符串包含"hdchina="，则保留该子字符串 + cookie += sub_str + ";" + + if "hdchina=" not in cookie: + logger.error(f"{site} 签到失败，Cookie已失效") + return False, '签到失败，Cookie已失效' + + site_cookie = cookie + # 获取页面html + html_res = RequestUtils(cookies=site_cookie, + ua=ua, + proxies=proxies + ).get_res(url="https://hdchina.org/index.php") + if not html_res or html_res.status_code != 200: + logger.error(f"{site} 签到失败，请检查站点连通性") + return False, '签到失败，请检查站点连通性' + + if "login.php" in html_res.text or "阻断页面" in html_res.text: + logger.error(f"{site} 签到失败，Cookie失效") + return False, '签到失败，Cookie失效' + + # 获取新返回的cookie进行签到 + site_cookie = ';'.join(['{}={}'.format(k, v) for k, v in html_res.cookies.get_dict().items()]) + + # 判断是否已签到 + html_res.encoding = "utf-8" + sign_status = self.sign_in_result(html_res=html_res.text, + regexs=self._sign_regex) + if sign_status: + logger.info(f"{site} 今日已签到") + return True, '今日已签到' + + # 没有签到则解析html + html = etree.HTML(html_res.text) + + if not html: + return False, '签到失败' + + # x_csrf + x_csrf = html.xpath("//meta[@name='x-csrf']/@content")[0] + if not x_csrf: + logger.error("{site} 签到失败，获取x-csrf失败") + return False, '签到失败' + logger.debug(f"获取到x-csrf {x_csrf}") + + # 签到 + data = { + 'csrf': x_csrf + } + sign_res = RequestUtils(cookies=site_cookie, + ua=ua, + proxies=proxies + ).post_res(url="https://hdchina.org/plugin_sign-in.php?cmd=signin", data=data) + if not sign_res or sign_res.status_code != 200: + logger.error(f"{site} 签到失败，签到接口请求失败") + return False, '签到失败，签到接口请求失败' + + sign_dict = json.loads(sign_res.text) + logger.debug(f"签到返回结果 {sign_dict}") + if sign_dict['state']: + # {'state': 'success', 'signindays': 10, 'integral': 20} + logger.info(f"{site} 签到成功") + return True, '签到成功' + else: + # {'state': False, 'msg': '不正确的CSRF / Incorrect CSRF token'} + logger.error(f"{site} 签到失败，不正确的CSRF / Incorrect CSRF token") + return False, '签到失败' diff --git a/plugins.v2/autosignin/sites/hdcity.py b/plugins.v2/autosignin/sites/hdcity.py new file mode 100644 index 0000000..229a523 --- /dev/null +++ b/plugins.v2/autosignin/sites/hdcity.py @@ -0,0 +1,66 @@ +from typing import Tuple + +from ruamel.yaml import CommentedMap + +from app.log import logger +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.string import StringUtils + + +class HDCity(_ISiteSigninHandler): + """ + 城市签到 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "hdcity.city" + + # 签到成功 + _success_text = '本次签到获得魅力' + # 重复签到 + _repeat_text = '已签到' + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if StringUtils.url_equal(url, cls.site_url) else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + site = site_info.get("name") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + proxy = site_info.get("proxy") + render = site_info.get("render") + + # 获取页面html + html_text = self.get_page_source(url='https://hdcity.city/sign', + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + if not html_text: + logger.error(f"{site} 签到失败，请检查站点连通性") + return False, '签到失败，请检查站点连通性' + + if "login" in html_text: + logger.error(f"{site} 签到失败，Cookie已失效") + return False, '签到失败，Cookie已失效' + + # 判断是否已签到 + # '已连续签到278天，此次签到您获得了100魔力值奖励!' + if self._success_text in html_text: + logger.info(f"{site} 签到成功") + return True, '签到成功' + if self._repeat_text in html_text: + logger.info(f"{site} 今日已签到") + return True, '今日已签到' + logger.error(f"{site} 签到失败，签到接口返回 {html_text}") + return False, '签到失败' diff --git a/plugins.v2/autosignin/sites/hdsky.py b/plugins.v2/autosignin/sites/hdsky.py new file mode 100644 index 0000000..0c3d844 --- /dev/null +++ b/plugins.v2/autosignin/sites/hdsky.py @@ -0,0 +1,138 @@ +import json +import time +from typing import Tuple + +from ruamel.yaml import CommentedMap + +from app.core.config import settings +from app.helper.ocr import OcrHelper +from app.log import logger +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.http import RequestUtils +from app.utils.string import StringUtils + + +class HDSky(_ISiteSigninHandler): + """ + 天空ocr签到 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "hdsky.me" + + # 已签到 + _sign_regex = ['已签到'] + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if StringUtils.url_equal(url, cls.site_url) else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + site = site_info.get("name") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + proxy = site_info.get("proxy") + render = site_info.get("render") + referer = site_info.get("url") + + # 判断今日是否已签到 + html_text = self.get_page_source(url='https://hdsky.me', + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + if not html_text: + logger.error(f"{site} 签到失败，请检查站点连通性") + return False, '签到失败，请检查站点连通性' + + if "login.php" in html_text: + logger.error(f"{site} 签到失败，Cookie已失效") + return False, '签到失败，Cookie已失效' + + sign_status = self.sign_in_result(html_res=html_text, + regexs=self._sign_regex) + if sign_status: + logger.info(f"{site} 今日已签到") + return True, '今日已签到' + + # 获取验证码请求，考虑到网络问题获取失败，多获取几次试试 + res_times = 0 + img_hash = None + while not img_hash and res_times <= 3: + image_res = RequestUtils(cookies=site_cookie, + ua=ua, + content_type='application/x-www-form-urlencoded; charset=UTF-8', + referer="https://hdsky.me/index.php", + accept_type="*/*", + proxies=settings.PROXY if proxy else None + ).post_res(url='https://hdsky.me/image_code_ajax.php', + data={'action': 'new'}) + if image_res and image_res.status_code == 200: + image_json = json.loads(image_res.text) + if image_json["success"]: + img_hash = image_json["code"] + break + res_times += 1 + logger.info(f"获取 {site} 验证码失败，正在进行重试，目前重试次数：{res_times}") + time.sleep(1) + + # 获取到二维码hash + if img_hash: + # 完整验证码url + img_get_url = 'https://hdsky.me/image.php?action=regimage&imagehash=%s' % img_hash + logger.info(f"获取到 {site} 验证码链接：{img_get_url}") + # ocr识别多次，获取6位验证码 + times = 0 + ocr_result = None + # 识别几次 + while times <= 3: + # ocr二维码识别 + ocr_result = OcrHelper().get_captcha_text(image_url=img_get_url, + cookie=site_cookie, + ua=ua) + logger.info(f"OCR识别 {site} 验证码：{ocr_result}") + if ocr_result: + if len(ocr_result) == 6: + logger.info(f"OCR识别 {site} 验证码成功：{ocr_result}") + break + times += 1 + logger.info(f"OCR识别 {site} 验证码失败，正在进行重试，目前重试次数：{times}") + time.sleep(1) + + if ocr_result: + # 组装请求参数 + data = { + 'action': 'showup', + 'imagehash': img_hash, + 'imagestring': ocr_result + } + # 访问签到链接 + res = RequestUtils(cookies=site_cookie, + ua=ua, + referer=referer, + proxies=settings.PROXY if proxy else None + ).post_res(url='https://hdsky.me/showup.php', data=data) + if res and res.status_code == 200: + if json.loads(res.text)["success"]: + logger.info(f"{site} 签到成功") + return True, '签到成功' + elif str(json.loads(res.text)["message"]) == "date_unmatch": + # 重复签到 + logger.warn(f"{site} 重复成功") + return True, '今日已签到' + elif str(json.loads(res.text)["message"]) == "invalid_imagehash": + # 验证码错误 + logger.warn(f"{site} 签到失败：验证码错误") + return False, '签到失败：验证码错误' + + logger.error(f'{site} 签到失败：未获取到验证码') + return False, '签到失败：未获取到验证码' diff --git a/plugins.v2/autosignin/sites/hdupt.py b/plugins.v2/autosignin/sites/hdupt.py new file mode 100644 index 0000000..470981d --- /dev/null +++ b/plugins.v2/autosignin/sites/hdupt.py @@ -0,0 +1,82 @@ +import re +from typing import Tuple + +from ruamel.yaml import CommentedMap + +from app.log import logger +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.string import StringUtils + + +class HDUpt(_ISiteSigninHandler): + """ + hdu签到 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "pt.hdupt.com" + + # 已签到 + _sign_regex = [''] + + # 签到成功 + _success_text = '本次签到获得魅力' + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if StringUtils.url_equal(url, cls.site_url) else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + site = site_info.get("name") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + proxy = site_info.get("proxy") + render = site_info.get("render") + + # 获取页面html + html_text = self.get_page_source(url='https://pt.hdupt.com', + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + if not html_text: + logger.error(f"{site} 签到失败，请检查站点连通性") + return False, '签到失败，请检查站点连通性' + + if "login.php" in html_text: + logger.error(f"{site} 签到失败，Cookie已失效") + return False, '签到失败，Cookie已失效' + + sign_status = self.sign_in_result(html_res=html_text, + regexs=self._sign_regex) + if sign_status: + logger.info(f"{site} 今日已签到") + return True, '今日已签到' + + # 签到 + html_text = self.get_page_source(url='https://pt.hdupt.com/added.php?action=qiandao', + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + if not html_text: + logger.error(f"{site} 签到失败，请检查站点连通性") + return False, '签到失败，请检查站点连通性' + + logger.debug(f"{site} 签到接口返回 {html_text}") + # 判断是否已签到 sign_res.text = ".23" + if len(list(map(int, re.findall(r"\d+", html_text)))) > 0: + logger.info(f"{site} 签到成功") + return True, '签到成功' + + logger.error(f"{site} 签到失败，签到接口返回 {html_text}") + return False, '签到失败' diff --git a/plugins.v2/autosignin/sites/mteam.py b/plugins.v2/autosignin/sites/mteam.py new file mode 100644 index 0000000..5db1ef1 --- /dev/null +++ b/plugins.v2/autosignin/sites/mteam.py @@ -0,0 +1,61 @@ +from typing import Tuple +from urllib.parse import urljoin + +from ruamel.yaml import CommentedMap + +from app.core.config import settings +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.http import RequestUtils +from app.utils.string import StringUtils + + +class MTorrent(_ISiteSigninHandler): + """ + m-team签到 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "m-team" + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if cls.site_url in url.split(".") else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作，馒头实际没有签到，非仿真模式下需要更新访问时间 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + headers = { + "Content-Type": "application/json", + "User-Agent": site_info.get("ua"), + "Accept": "application/json, text/plain, */*", + "Authorization": site_info.get("token") + } + url = site_info.get('url') + domain = StringUtils.get_url_domain(url) + # 更新最后访问时间 + res = RequestUtils(headers=headers, + timeout=60, + proxies=settings.PROXY if site_info.get("proxy") else None, + referer=f"{url}index" + ).post_res(url=f"https://api.{domain}/api/member/updateLastBrowse") + if res: + return True, "模拟登录成功" + elif res is not None: + return False, f"模拟登录失败，状态码：{res.status_code}" + else: + return False, "模拟登录失败，无法打开网站" + + def login(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行登录操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 登录结果信息 + """ + return self.signin(site_info) diff --git a/plugins.v2/autosignin/sites/nexushd.py b/plugins.v2/autosignin/sites/nexushd.py new file mode 100644 index 0000000..78941c0 --- /dev/null +++ b/plugins.v2/autosignin/sites/nexushd.py @@ -0,0 +1,70 @@ +from typing import Tuple + +from ruamel.yaml import CommentedMap + +from app.core.config import settings +from app.log import logger +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.http import RequestUtils +from app.utils.string import StringUtils + + +class NexusHD(_ISiteSigninHandler): + """ + NexusHD签到 + """ + + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "v6.nexushd.org" + + # 签到成功 + _success_text = "本次签到获得" + _repeat_text = "你今天已经签到过了" + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if StringUtils.url_equal(url, cls.site_url) else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + site = site_info.get("name") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + proxies = settings.PROXY if site_info.get("proxy") else None + + # 获取页面html + data = { + 'action': 'post', + 'content': '' + } + html_res = RequestUtils(cookies=site_cookie, + ua=ua, + proxies=proxies + ).post_res(url="https://v6.nexushd.org/signin.php", data=data) + if not html_res or html_res.status_code != 200: + logger.error(f"{site} 签到失败，请检查站点连通性") + return False, '签到失败，请检查站点连通性' + + if "login.php" in html_res.text: + logger.error(f"{site} 签到失败，Cookie已失效") + return False, '签到失败，Cookie已失效' + + # 判断是否已签到 + # '已连续签到278天，此次签到您获得了100魔力值奖励!' + if self._success_text in html_res.text: + logger.info(f"{site} 签到成功") + return True, '签到成功' + if self._repeat_text in html_res.text: + logger.info(f"{site} 今日已签到") + return True, '今日已签到' + logger.error(f"{site} 签到失败，签到接口返回 {html_res.text}") + return False, '签到失败' diff --git a/plugins.v2/autosignin/sites/opencd.py b/plugins.v2/autosignin/sites/opencd.py new file mode 100644 index 0000000..1f8d0c1 --- /dev/null +++ b/plugins.v2/autosignin/sites/opencd.py @@ -0,0 +1,132 @@ +import json +import time +from typing import Tuple + +from lxml import etree +from ruamel.yaml import CommentedMap + +from app.core.config import settings +from app.helper.ocr import OcrHelper +from app.log import logger +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.http import RequestUtils +from app.utils.string import StringUtils + + +class Opencd(_ISiteSigninHandler): + """ + 皇后ocr签到 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "open.cd" + + # 已签到 + _repeat_text = "/plugin_sign-in.php?cmd=show-log" + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if StringUtils.url_equal(url, cls.site_url) else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + site = site_info.get("name") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + proxy = site_info.get("proxy") + render = site_info.get("render") + + # 判断今日是否已签到 + html_text = self.get_page_source(url='https://www.open.cd', + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + if not html_text: + logger.error(f"{site} 签到失败，请检查站点连通性") + return False, '签到失败，请检查站点连通性' + + if "login.php" in html_text: + logger.error(f"{site} 签到失败，Cookie已失效") + return False, '签到失败，Cookie已失效' + + if self._repeat_text in html_text: + logger.info(f"{site} 今日已签到") + return True, '今日已签到' + + # 获取签到参数 + html_text = self.get_page_source(url='https://www.open.cd/plugin_sign-in.php', + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + if not html_text: + logger.error(f"{site} 签到失败，请检查站点连通性") + return False, '签到失败，请检查站点连通性' + + # 没有签到则解析html + html = etree.HTML(html_text) + if not html: + return False, '签到失败' + + # 签到参数 + img_url = html.xpath('//form[@id="frmSignin"]//img/@src')[0] + img_hash = html.xpath('//form[@id="frmSignin"]//input[@name="imagehash"]/@value')[0] + if not img_url or not img_hash: + logger.error(f"{site} 签到失败，获取签到参数失败") + return False, '签到失败，获取签到参数失败' + + # 完整验证码url + img_get_url = 'https://www.open.cd/%s' % img_url + logger.debug(f"{site} 获取到{site}验证码链接 {img_get_url}") + + # ocr识别多次，获取6位验证码 + times = 0 + ocr_result = None + # 识别几次 + while times <= 3: + # ocr二维码识别 + ocr_result = OcrHelper().get_captcha_text(image_url=img_get_url, + cookie=site_cookie, + ua=ua) + logger.debug(f"ocr识别{site}验证码 {ocr_result}") + if ocr_result: + if len(ocr_result) == 6: + logger.info(f"ocr识别{site}验证码成功 {ocr_result}") + break + times += 1 + logger.debug(f"ocr识别{site}验证码失败，正在进行重试，目前重试次数 {times}") + time.sleep(1) + + if ocr_result: + # 组装请求参数 + data = { + 'imagehash': img_hash, + 'imagestring': ocr_result + } + # 访问签到链接 + sign_res = RequestUtils(cookies=site_cookie, + ua=ua, + proxies=settings.PROXY if proxy else None + ).post_res(url='https://www.open.cd/plugin_sign-in.php?cmd=signin', data=data) + if sign_res and sign_res.status_code == 200: + logger.debug(f"sign_res返回 {sign_res.text}") + # sign_res.text = '{"state":"success","signindays":"0","integral":"10"}' + sign_dict = json.loads(sign_res.text) + if sign_dict['state']: + logger.info(f"{site} 签到成功") + return True, '签到成功' + else: + logger.error(f"{site} 签到失败，签到接口返回 {sign_dict}") + return False, '签到失败' + + logger.error(f'{site} 签到失败：未获取到验证码') + return False, '签到失败：未获取到验证码' diff --git a/plugins.v2/autosignin/sites/pterclub.py b/plugins.v2/autosignin/sites/pterclub.py new file mode 100644 index 0000000..4047272 --- /dev/null +++ b/plugins.v2/autosignin/sites/pterclub.py @@ -0,0 +1,65 @@ +import json +from typing import Tuple + +from ruamel.yaml import CommentedMap + +from app.log import logger +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.string import StringUtils + + +class PTerClub(_ISiteSigninHandler): + """ + 猫签到 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "pterclub.com" + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if StringUtils.url_equal(url, cls.site_url) else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + site = site_info.get("name") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + proxy = site_info.get("proxy") + render = site_info.get("render") + + # 签到 + html_text = self.get_page_source(url='https://pterclub.com/attendance-ajax.php', + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + if not html_text: + logger.error(f"{site} 签到失败，请检查站点连通性") + return False, '签到失败，请检查站点连通性' + + if "login.php" in html_text: + logger.error(f"{site} 签到失败，Cookie已失效") + return False, '签到失败，Cookie已失效' + try: + sign_dict = json.loads(html_text) + except Exception as e: + logger.error(f"{site} 签到失败，签到接口返回数据异常，错误信息：{str(e)}") + return False, '签到失败，签到接口返回数据异常' + if sign_dict['status'] == '1': + # {"status":"1","data":" (签到已成功300)","message":"

这是您的第237次签到， + # 已连续签到237天。

本次签到获得300克猫粮。

"} + logger.info(f"{site} 签到成功") + return True, '签到成功' + else: + # {"status":"0","data":"抱歉","message":"您今天已经签到过了，请勿重复刷新。"} + logger.info(f"{site} 今日已签到") + return True, '今日已签到' diff --git a/plugins.v2/autosignin/sites/pttime.py b/plugins.v2/autosignin/sites/pttime.py new file mode 100644 index 0000000..6c766d2 --- /dev/null +++ b/plugins.v2/autosignin/sites/pttime.py @@ -0,0 +1,64 @@ +from typing import Tuple + +from ruamel.yaml import CommentedMap + +from app.log import logger +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.string import StringUtils + + +class PTTime(_ISiteSigninHandler): + """ + PT时间签到 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "pttime.org" + + # 签到成功 + _succeed_regex = ['签到成功'] + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if StringUtils.url_equal(url, cls.site_url) else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + site = site_info.get("name") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + proxy = site_info.get("proxy") + render = site_info.get("render") + + # 签到 + # 签到返回：签到成功 + html_text = self.get_page_source(url='https://www.pttime.org/attendance.php', + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + + if not html_text: + logger.error(f"{site} 签到失败，请检查站点连通性") + return False, '签到失败，请检查站点连通性' + + if "login.php" in html_text: + logger.error(f"{site} 签到失败，Cookie已失效") + return False, '签到失败，Cookie已失效' + + sign_status = self.sign_in_result(html_res=html_text, + regexs=self._succeed_regex) + if sign_status: + logger.info(f"{site} 签到成功") + return True, '签到成功' + + logger.error(f"{site} 签到失败，签到接口返回 {html_text}") + return False, '签到失败' diff --git a/plugins.v2/autosignin/sites/tjupt.py b/plugins.v2/autosignin/sites/tjupt.py new file mode 100644 index 0000000..4a20f84 --- /dev/null +++ b/plugins.v2/autosignin/sites/tjupt.py @@ -0,0 +1,274 @@ +import json +import os +import time +from io import BytesIO +from typing import Tuple + +from PIL import Image +from lxml import etree +from ruamel.yaml import CommentedMap + +from app.core.config import settings +from app.log import logger +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.http import RequestUtils +from app.utils.string import StringUtils + + +class Tjupt(_ISiteSigninHandler): + """ + 北洋签到 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "tjupt.org" + + # 签到地址 + _sign_in_url = 'https://www.tjupt.org/attendance.php' + + # 已签到 + _sign_regex = ['今日已签到'] + + # 签到成功 + _succeed_regex = ['这是您的首次签到，本次签到获得\\d+个魔力值。', + '签到成功，这是您的第\\d+次签到，已连续签到\\d+天，本次签到获得\\d+个魔力值。', + '重新签到成功，本次签到获得\\d+个魔力值'] + + # 存储正确的答案，后续可直接查 + _answer_path = settings.TEMP_PATH / "signin/" + _answer_file = _answer_path / "tjupt.json" + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if StringUtils.url_equal(url, cls.site_url) else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + site = site_info.get("name") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + proxy = site_info.get("proxy") + render = site_info.get("render") + + # 创建正确答案存储目录 + if not os.path.exists(os.path.dirname(self._answer_file)): + os.makedirs(os.path.dirname(self._answer_file)) + + # 获取北洋签到页面html + html_text = self.get_page_source(url=self._sign_in_url, + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + + # 获取签到后返回html，判断是否签到成功 + if not html_text: + logger.error(f"{site} 签到失败，请检查站点连通性") + return False, '签到失败，请检查站点连通性' + + if "login.php" in html_text: + logger.error(f"{site} 签到失败，Cookie已失效") + return False, '签到失败，Cookie已失效' + + sign_status = self.sign_in_result(html_res=html_text, + regexs=self._sign_regex) + if sign_status: + logger.info(f"{site} 今日已签到") + return True, '今日已签到' + + # 没有签到则解析html + html = etree.HTML(html_text) + if not html: + return False, '签到失败' + img_url = html.xpath('//table[@class="captcha"]//img/@src')[0] + + if not img_url: + logger.error(f"{site} 签到失败，未获取到签到图片") + return False, '签到失败，未获取到签到图片' + + # 签到图片 + img_url = "https://www.tjupt.org" + img_url + logger.info(f"获取到签到图片 {img_url}") + # 获取签到图片hash + captcha_img_res = RequestUtils(cookies=site_cookie, + ua=ua, + proxies=settings.PROXY if proxy else None + ).get_res(url=img_url) + if not captcha_img_res or captcha_img_res.status_code != 200: + logger.error(f"{site} 签到图片 {img_url} 请求失败") + return False, '签到失败，未获取到签到图片' + captcha_img = Image.open(BytesIO(captcha_img_res.content)) + captcha_img_hash = self._tohash(captcha_img) + logger.debug(f"签到图片hash {captcha_img_hash}") + + # 签到答案选项 + values = html.xpath("//input[@name='answer']/@value") + options = html.xpath("//input[@name='answer']/following-sibling::text()") + + if not values or not options: + logger.error(f"{site} 签到失败，未获取到答案选项") + return False, '签到失败，未获取到答案选项' + + # value+选项 + answers = list(zip(values, options)) + logger.debug(f"获取到所有签到选项 {answers}") + + # 查询已有答案 + exits_answers = {} + try: + with open(self._answer_file, 'r') as f: + json_str = f.read() + exits_answers = json.loads(json_str) + # 查询本地本次验证码hash答案 + captcha_answer = exits_answers[captcha_img_hash] + + # 本地存在本次hash对应的正确答案再遍历查询 + if captcha_answer: + for value, answer in answers: + if str(captcha_answer) == str(answer): + # 确实是答案 + return self.__signin(answer=value, + site_cookie=site_cookie, + ua=ua, + proxy=proxy, + site=site) + except (FileNotFoundError, IOError, OSError) as e: + logger.debug(f"查询本地已知答案失败：{str(e)}，继续请求豆瓣查询") + + # 本地不存在正确答案则请求豆瓣查询匹配 + for value, answer in answers: + if answer: + # 豆瓣检索 + db_res = RequestUtils().get_res(url=f'https://movie.douban.com/j/subject_suggest?q={answer}') + if not db_res or db_res.status_code != 200: + logger.debug(f"签到选项 {answer} 未查询到豆瓣数据") + continue + + # 豆瓣返回结果 + db_answers = json.loads(db_res.text) + if not isinstance(db_answers, list): + db_answers = [db_answers] + + if len(db_answers) == 0: + logger.debug(f"签到选项 {answer} 查询到豆瓣数据为空") + + for db_answer in db_answers: + answer_img_url = db_answer['img'] + + # 获取答案hash + answer_img_res = RequestUtils(referer="https://movie.douban.com").get_res(url=answer_img_url) + if not answer_img_res or answer_img_res.status_code != 200: + logger.debug(f"签到答案 {answer} {answer_img_url} 请求失败") + continue + + answer_img = Image.open(BytesIO(answer_img_res.content)) + answer_img_hash = self._tohash(answer_img) + logger.debug(f"签到答案图片hash {answer} {answer_img_hash}") + + # 获取选项图片与签到图片相似度，大于0.9默认是正确答案 + score = self._comparehash(captcha_img_hash, answer_img_hash) + logger.info(f"签到图片与选项 {answer} 豆瓣图片相似度 {score}") + if score > 0.9: + # 确实是答案 + return self.__signin(answer=value, + site_cookie=site_cookie, + ua=ua, + proxy=proxy, + site=site, + exits_answers=exits_answers, + captcha_img_hash=captcha_img_hash) + + # 间隔5s，防止请求太频繁被豆瓣屏蔽ip + time.sleep(5) + logger.error(f"豆瓣图片匹配，未获取到匹配答案") + + # 没有匹配签到成功，则签到失败 + return False, '签到失败，未获取到匹配答案' + + def __signin(self, answer, site_cookie, ua, proxy, site, exits_answers=None, captcha_img_hash=None): + """ + 签到请求 + """ + data = { + 'answer': answer, + 'submit': '提交' + } + logger.debug(f"提交data {data}") + sign_in_res = RequestUtils(cookies=site_cookie, + ua=ua, + proxies=settings.PROXY if proxy else None + ).post_res(url=self._sign_in_url, data=data) + if not sign_in_res or sign_in_res.status_code != 200: + logger.error(f"{site} 签到失败，签到接口请求失败") + return False, '签到失败，签到接口请求失败' + + # 获取签到后返回html，判断是否签到成功 + sign_status = self.sign_in_result(html_res=sign_in_res.text, + regexs=self._succeed_regex) + if sign_status: + logger.info(f"签到成功") + if exits_answers and captcha_img_hash: + # 签到成功写入本地文件 + self.__write_local_answer(exits_answers=exits_answers or {}, + captcha_img_hash=captcha_img_hash, + answer=answer) + return True, '签到成功' + else: + logger.error(f"{site} 签到失败，请到页面查看") + return False, '签到失败，请到页面查看' + + def __write_local_answer(self, exits_answers, captcha_img_hash, answer): + """ + 签到成功写入本地文件 + """ + try: + exits_answers[captcha_img_hash] = answer + # 序列化数据 + formatted_data = json.dumps(exits_answers, indent=4) + with open(self._answer_file, 'w') as f: + f.write(formatted_data) + except (FileNotFoundError, IOError, OSError) as e: + logger.debug(f"签到成功写入本地文件失败：{str(e)}") + + @staticmethod + def _tohash(img, shape=(10, 10)): + """ + 获取图片hash + """ + img = img.resize(shape) + gray = img.convert('L') + s = 0 + hash_str = '' + for i in range(shape[1]): + for j in range(shape[0]): + s = s + gray.getpixel((j, i)) + avg = s / (shape[0] * shape[1]) + for i in range(shape[1]): + for j in range(shape[0]): + if gray.getpixel((j, i)) > avg: + hash_str = hash_str + '1' + else: + hash_str = hash_str + '0' + return hash_str + + @staticmethod + def _comparehash(hash1, hash2, shape=(10, 10)): + """ + 比较图片hash + 返回相似度 + """ + n = 0 + if len(hash1) != len(hash2): + return -1 + for i in range(len(hash1)): + if hash1[i] == hash2[i]: + n = n + 1 + return n / (shape[0] * shape[1]) diff --git a/plugins.v2/autosignin/sites/ttg.py b/plugins.v2/autosignin/sites/ttg.py new file mode 100644 index 0000000..d3470a6 --- /dev/null +++ b/plugins.v2/autosignin/sites/ttg.py @@ -0,0 +1,97 @@ +import re +from typing import Tuple + +from ruamel.yaml import CommentedMap + +from app.core.config import settings +from app.log import logger +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.http import RequestUtils +from app.utils.string import StringUtils + + +class TTG(_ISiteSigninHandler): + """ + TTG签到 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "totheglory.im" + + # 已签到 + _sign_regex = ['已签到'] + _sign_text = '亲，您今天已签到过，不要太贪哦' + + # 签到成功 + _success_text = '您已连续签到' + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if StringUtils.url_equal(url, cls.site_url) else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + site = site_info.get("name") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + proxy = site_info.get("proxy") + render = site_info.get("render") + + # 获取页面html + html_text = self.get_page_source(url="https://totheglory.im", + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + if not html_text: + logger.error(f"{site} 签到失败，请检查站点连通性") + return False, '签到失败，请检查站点连通性' + + if "login.php" in html_text: + logger.error(f"{site} 签到失败，Cookie已失效") + return False, '签到失败，Cookie已失效' + + # 判断是否已签到 + sign_status = self.sign_in_result(html_res=html_text, + regexs=self._sign_regex) + if sign_status: + logger.info(f"{site} 今日已签到") + return True, '今日已签到' + + # 获取签到参数 + signed_timestamp = re.search('(?<=signed_timestamp: ")\\d{10}', html_text).group() + signed_token = re.search('(?<=signed_token: ").*(?=")', html_text).group() + logger.debug(f"signed_timestamp={signed_timestamp} signed_token={signed_token}") + + data = { + 'signed_timestamp': signed_timestamp, + 'signed_token': signed_token + } + # 签到 + sign_res = RequestUtils(cookies=site_cookie, + ua=ua, + proxies=settings.PROXY if proxy else None + ).post_res(url="https://totheglory.im/signed.php", + data=data) + if not sign_res or sign_res.status_code != 200: + logger.error(f"{site} 签到失败，签到接口请求失败") + return False, '签到失败，签到接口请求失败' + + sign_res.encoding = "utf-8" + if self._success_text in sign_res.text: + logger.info(f"{site} 签到成功") + return True, '签到成功' + if self._sign_text in sign_res.text: + logger.info(f"{site} 今日已签到") + return True, '今日已签到' + + logger.error(f"{site} 签到失败，未知原因") + return False, '签到失败，未知原因' diff --git a/plugins.v2/autosignin/sites/u2.py b/plugins.v2/autosignin/sites/u2.py new file mode 100644 index 0000000..2c45c2c --- /dev/null +++ b/plugins.v2/autosignin/sites/u2.py @@ -0,0 +1,123 @@ +import datetime +import random +import re +from typing import Tuple + +from lxml import etree +from ruamel.yaml import CommentedMap + +from app.core.config import settings +from app.log import logger +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.http import RequestUtils +from app.utils.string import StringUtils + + +class U2(_ISiteSigninHandler): + """ + U2签到 随机 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "u2.dmhy.org" + + # 已签到 + _sign_regex = ['已签到', + 'Show Up', + 'Показать', + '已簽到', + '已簽到'] + + # 签到成功 + _success_text = "window.location.href = 'showup.php';" + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if StringUtils.url_equal(url, cls.site_url) else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + site = site_info.get("name") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + proxy = site_info.get("proxy") + render = site_info.get("render") + + now = datetime.datetime.now() + # 判断当前时间是否小于9点 + if now.hour < 9: + logger.error(f"{site} 签到失败，9点前不签到") + return False, '签到失败，9点前不签到' + + # 获取页面html + html_text = self.get_page_source(url="https://u2.dmhy.org/showup.php", + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + if not html_text: + logger.error(f"{site} 签到失败，请检查站点连通性") + return False, '签到失败，请检查站点连通性' + + if "login.php" in html_text: + logger.error(f"{site} 签到失败，Cookie已失效") + return False, '签到失败，Cookie已失效' + + # 判断是否已签到 + sign_status = self.sign_in_result(html_res=html_text, + regexs=self._sign_regex) + if sign_status: + logger.info(f"{site} 今日已签到") + return True, '今日已签到' + + # 没有签到则解析html + html = etree.HTML(html_text) + + if not html: + return False, '签到失败' + + # 获取签到参数 + req = html.xpath("//form//td/input[@name='req']/@value")[0] + hash_str = html.xpath("//form//td/input[@name='hash']/@value")[0] + form = html.xpath("//form//td/input[@name='form']/@value")[0] + submit_name = html.xpath("//form//td/input[@type='submit']/@name") + submit_value = html.xpath("//form//td/input[@type='submit']/@value") + if not re or not hash_str or not form or not submit_name or not submit_value: + logger.error("{site} 签到失败，未获取到相关签到参数") + return False, '签到失败' + + # 随机一个答案 + answer_num = random.randint(0, 3) + data = { + 'req': req, + 'hash': hash_str, + 'form': form, + 'message': '一切随缘~', + submit_name[answer_num]: submit_value[answer_num] + } + # 签到 + sign_res = RequestUtils(cookies=site_cookie, + ua=ua, + proxies=settings.PROXY if proxy else None + ).post_res(url="https://u2.dmhy.org/showup.php?action=show", + data=data) + if not sign_res or sign_res.status_code != 200: + logger.error(f"{site} 签到失败，签到接口请求失败") + return False, '签到失败，签到接口请求失败' + + # 判断是否签到成功 + # sign_res.text = "" + if self._success_text in sign_res.text: + logger.info(f"{site} 签到成功") + return True, '签到成功' + else: + logger.error(f"{site} 签到失败，未知原因") + return False, '签到失败，未知原因' diff --git a/plugins.v2/autosignin/sites/yema.py b/plugins.v2/autosignin/sites/yema.py new file mode 100644 index 0000000..879611f --- /dev/null +++ b/plugins.v2/autosignin/sites/yema.py @@ -0,0 +1,78 @@ +from typing import Tuple +from urllib.parse import urljoin + +from ruamel.yaml import CommentedMap + +from app.core.config import settings +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.http import RequestUtils + + +class YemaPT(_ISiteSigninHandler): + """ + YemaPT 签到 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "yemapt.org" + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if cls.site_url in url else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + headers = { + "Content-Type": "application/json", + "User-Agent": site_info.get("ua"), + "Accept": "application/json, text/plain, */*", + } + # 获取用户信息，更新最后访问时间 + res = (RequestUtils(headers=headers, + timeout=15, + cookies=site_info.get("cookie"), + proxies=settings.PROXY if site_info.get("proxy") else None, + referer=site_info.get('url') + ).get_res(urljoin(site_info.get('url'), "api/consumer/checkIn"))) + + if res and res.json().get("success"): + return True, "签到成功" + elif res is not None: + return False, f"签到失败，签到结果：{res.json().get('errorMessage')}" + else: + return False, "签到失败，无法打开网站" + + def login(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行登录操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 登录结果信息 + """ + + headers = { + "Content-Type": "application/json", + "User-Agent": site_info.get("ua"), + "Accept": "application/json, text/plain, */*", + } + # 获取用户信息，更新最后访问时间 + res = (RequestUtils(headers=headers, + timeout=15, + cookies=site_info.get("cookie"), + proxies=settings.PROXY if site_info.get("proxy") else None, + referer=site_info.get('url') + ).get_res(urljoin(site_info.get('url'), "api/user/profile"))) + + if res and res.json().get("success"): + return True, "模拟登录成功" + elif res is not None: + return False, f"模拟登录失败，状态码：{res.status_code}" + else: + return False, "模拟登录失败，无法打开网站" diff --git a/plugins.v2/autosignin/sites/zhuque.py b/plugins.v2/autosignin/sites/zhuque.py new file mode 100644 index 0000000..f3375f5 --- /dev/null +++ b/plugins.v2/autosignin/sites/zhuque.py @@ -0,0 +1,88 @@ +import json +from typing import Tuple + +from lxml import etree +from ruamel.yaml import CommentedMap + +from app.core.config import settings +from app.log import logger +from app.plugins.autosignin.sites import _ISiteSigninHandler +from app.utils.http import RequestUtils +from app.utils.string import StringUtils + + +class ZhuQue(_ISiteSigninHandler): + """ + ZHUQUE签到 + """ + # 匹配的站点Url，每一个实现类都需要设置为自己的站点Url + site_url = "zhuque.in" + + @classmethod + def match(cls, url: str) -> bool: + """ + 根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可 + :param url: 站点Url + :return: 是否匹配，如匹配则会调用该类的signin方法 + """ + return True if StringUtils.url_equal(url, cls.site_url) else False + + def signin(self, site_info: CommentedMap) -> Tuple[bool, str]: + """ + 执行签到操作 + :param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息 + :return: 签到结果信息 + """ + site = site_info.get("name") + site_cookie = site_info.get("cookie") + ua = site_info.get("ua") + proxy = site_info.get("proxy") + render = site_info.get("render") + + # 获取页面html + html_text = self.get_page_source(url="https://zhuque.in", + cookie=site_cookie, + ua=ua, + proxy=proxy, + render=render) + if not html_text: + logger.error(f"{site} 模拟登录失败，请检查站点连通性") + return False, '模拟登录失败，请检查站点连通性' + + if "login.php" in html_text: + logger.error(f"{site} 模拟登录失败，Cookie已失效") + return False, '模拟登录失败，Cookie已失效' + + html = etree.HTML(html_text) + + if not html: + return False, '模拟登录失败' + + # 释放技能 + msg = '失败' + x_csrf_token = html.xpath("//meta[@name='x-csrf-token']/@content")[0] + if x_csrf_token: + data = { + "all": 1, + "resetModal": "true" + } + headers = { + "x-csrf-token": str(x_csrf_token), + "Content-Type": "application/json; charset=utf-8", + "User-Agent": ua + } + skill_res = RequestUtils(cookies=site_cookie, + headers=headers, + proxies=settings.PROXY if proxy else None + ).post_res(url="https://zhuque.in/api/gaming/fireGenshinCharacterMagic", json=data) + if not skill_res or skill_res.status_code != 200: + logger.error(f"模拟登录失败，释放技能失败") + + # '{"status":200,"data":{"code":"FIRE_GENSHIN_CHARACTER_MAGIC_SUCCESS","bonus":0}}' + skill_dict = json.loads(skill_res.text) + if skill_dict['status'] == 200: + bonus = int(skill_dict['data']['bonus']) + msg = f'成功，获得{bonus}魔力' + + logger.info(f'【{site}】模拟登录成功，技能释放{msg}') + return True, f'模拟登录成功，技能释放{msg}'