1.购买服务器阿里云:服务器购买地址https://t.aliyun.com/U/DT4XYh若失效,可用地址
阿里云:
服务器购买地址
https://t.aliyun.com/U/DT4XYh若失效,可用地址
https://www.aliyun.com/activity/wuying/dj?source=5176.29345612&userCode=49hts92d腾讯云:
https://curl.qcloud.com/wJpWmSfU若失效,可用地址
https://cloud.tencent.com/act/cps/redirect?redirect=2446&cps_key=ad201ee2ef3b771157f72ee5464b1fea&from=console华为云
https://activity.huaweicloud.com/cps.html?fromacct=64b5cf7cc11b4840bb4ed2ea0b2f4468&utm_source=V1g3MDY4NTY=&utm_medium=cps&utm_campaign=2019052.部署教程
3.代码如下
import requestsimport reimport osimport jsonimport timeimport loggingimport sysimport subprocessfrom typing import Optional, Tuple# 配置日志logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')logger = logging.getLogger(__name__)class EnshanSign:def __init__(self):# 从环境变量获取配置self.enshanck = os.getenv("enshanck")# 请求配置self.headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36","Cookie": self.enshanck if self.enshanck else "","Referer": "https://www.right.com.cn/FORUM/","Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8","Accept-Language": "zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2","Accept-Encoding": "gzip, deflate, br","Connection": "keep-alive","Upgrade-Insecure-Requests": "1"}# 请求会话self.session = requests.Session()self.session.headers.update(self.headers)# 最大重试次数self.max_retries = 3self.retry_delay = 5 # 重试延迟(秒)def check_config(self) -> bool:"""检查必要配置是否存在"""if not self.enshanck:logger.error("未找到恩山Cookie配置,请设置环境变量: enshanck")return Falsereturn Truedef get_enshan_info(self) -> Tuple[Optional[str], Optional[str]]:"""获取恩山论坛的积分信息Returns:Tuple[coin, point]: 恩山币和积分"""url = 'https://www.right.com.cn/FORUM/home.php?mod=spacecp&ac=credit&showcredit=1'for attempt in range(self.max_retries):try:response = self.session.get(url, timeout=10)response.raise_for_status() # 检查HTTP状态码# 使用更健壮的正则表达式匹配coin_pattern = r'恩山币[^:]*:\s*</em>([^<&]+)'point_pattern = r'积分[^:]*:\s*</em>([^<&]+)'coin_match = re.search(coin_pattern, response.text, re.IGNORECASE)point_match = re.search(point_pattern, response.text, re.IGNORECASE)if coin_match and point_match:coin = coin_match.group(1).strip()point = point_match.group(1).strip()return coin, pointelse:logger.warning(f"第{attempt+1}次尝试: 未找到积分信息,可能是页面结构变化或Cookie失效")except requests.exceptions.RequestException as e:logger.error(f"第{attempt+1}次请求失败: {str(e)}")# 如果不是最后一次尝试,等待后重试if attempt < self.max_retries - 1:time.sleep(self.retry_delay)return None, Nonedef send_notify(self, title: str, content: str):"""使用青龙面板的notify.py发送通知Args:title: 通知标题content: 通知内容"""try:# 查找notify.py文件路径script_dir = os.path.dirname(os.path.abspath(__file__))notify_py_path = os.path.join(script_dir, "notify.py")if not os.path.exists(notify_py_path):# 尝试在上级目录查找parent_dir = os.path.dirname(script_dir)notify_py_path = os.path.join(parent_dir, "notify.py")if not os.path.exists(notify_py_path):logger.warning("未找到notify.py文件,无法发送通知")return# 使用Python直接导入notify.py模块并调用发送函数# 首先将notify.py所在目录添加到Python路径notify_dir = os.path.dirname(notify_py_path)if notify_dir not in sys.path:sys.path.insert(0, notify_dir)# 尝试导入notify模块try:import notify# 调用notify的发送函数notify.send(title, content)logger.info("使用notify.py发送通知成功")except ImportError as e:logger.error(f"导入notify模块失败: {str(e)}")# 如果导入失败,尝试使用命令行方式调用self._fallback_notify(notify_py_path, title, content)except Exception as e:logger.error(f"发送通知时发生错误: {str(e)}")def _fallback_notify(self, notify_py_path: str, title: str, content: str):"""备用的通知发送方式(通过命令行调用)Args:notify_py_path: notify.py文件路径title: 通知标题content: 通知内容"""try:# 创建临时数据文件data = {"title": title,"content": content}script_dir = os.path.dirname(os.path.abspath(__file__))temp_file = os.path.join(script_dir, "temp_notify_data.json")with open(temp_file, 'w', encoding='utf-8') as f:json.dump(data, f, ensure_ascii=False)# 使用Python执行notify.py并传递参数cmd = f"python3 {notify_py_path} '{title}' '{content}'"result = subprocess.run(cmd, shell=True, capture_output=True, text=True, cwd=script_dir)# 清理临时文件if os.path.exists(temp_file):os.remove(temp_file)if result.returncode == 0:logger.info("通过命令行调用notify.py发送通知成功")else:logger.error(f"通过命令行调用notify.py发送通知失败: {result.stderr}")except Exception as e:logger.error(f"备用通知方式也失败了: {str(e)}")def run(self):"""主运行逻辑"""logger.info("开始恩山签到任务")# 检查配置if not self.check_config():return# 获取积分信息coin, point = self.get_enshan_info()if coin is not None and point is not None:title = "恩山签到成功"content = f"恩山币:{coin}\n积分:{point}"logger.info(f"获取信息成功: {content}")else:title = "恩山签到失败"content = "获取恩山积分信息失败,请检查Cookie是否有效"logger.error(content)# 发送通知self.send_notify(title, content)if __name__ == "__main__":sign = EnshanSign()sign.run()
解析
该脚本为恩山论坛自动签到脚本,主要作用包括:
通过携带 Cookie(环境变量
enshanck) 访问恩山论坛"积分页面",抓取当前账号的"恩山币"和"积分"数值。将结果通过青龙面板的
推送通知(成功/失败都推)。
运行流程
从环境变量读取
enshanck,设置到请求头Cookie。用会话请求积分页
home.php?mod=spacecp&ac=credit&showcredit=1用正则从返回的 HTML 中提取"恩山币""积分"。
组织消息并调用
notify.py的send(title, content)推送结果。请求失败或解析不到就重试(最多 3 次),仍失败则推送失败通知。
主要方法作用
__init__(self)读取环境变量
enshanck。组装通用请求头(含 UA、Cookie、Referer 等)。
创建
requests.Session()会话并设置默认头。设置重试参数(
max_retries=3,retry_delay=5s)。check_config(self) -> bool检查必须的 Cookie 是否存在;没有则记录错误并中止后续流程。
get_enshan_info(self) -> Tuple[Optional[str], Optional[str]]访问积分页,最多重试 3 次。
用正则匹配页面里的"恩山币""积分"数值并返回
(coin, point)。失败则返回
(None, None)并输出原因(网络错误、页面结构变化、Cookie 失效等)。send_notify(self, title: str, content: str)在脚本目录或上级目录查找
notify.py,优先模块导入调用notify.send(...)。若导入失败,调用
_fallback_notify(...)走命令行兜底。_fallback_notify(self, notify_py_path: str, title: str, content: str)以命令行
python3 notify.py 'title' 'content'的方式触发通知,作为备选方案。过程里简单写/删一个临时文件(当前逻辑并未真正使用该临时文件的内容)。
run(self)总控:日志开场 →
check_config→get_enshan_info→ 拼装通知内容 →send_notify。
依赖与配置
依赖:
requests(必需)。
安装:pip3 install requests配置:在青龙/环境中设置
export enshanck='你的恩山Cookie'
注意:
本文部分变量已做脱敏处理,仅用于测试和学习研究,禁止用于商业用途,不能保证其合法性,准确性,完整性和有效性,请根据情况自行判断。技术层面需要提供帮助,可以通过打赏的方式进行探讨。
【相关文章】
没有评论:
发表评论