utils.py

from abc import ABC, abstractmethod
from datetime import datetime, timedelta
from urllib.parse import urlparse
import base64
import os
from cryptography.fernet import Fernet


import apprise
from loguru import logger

logger.add("logs/cross_bj.log", rotation="100 MB", retention="30 days")

class SendMessage(ABC):
    @abstractmethod
    def send(self, title, msg):
        pass


class AppriseNotifier(SendMessage):
    def __init__(self, urls=None):
        """
        初始化Apprise通知器
        
        Args:
            urls: 推送服务URL列表，支持多种推送方式
            详见: https://github.com/caronc/apprise
                例如: [
                    "barks://api.day.app/your_bark_token",
                    "tgram://bottoken/ChatID",
                    "email://user:pass@smtp.gmail.com:587",
                    "slack://TokenA/TokenB/TokenC/",
                    "discord://webhook_id/webhook_token",
                    "webhook://your_webhook_url"
                ]
        """
        self.apobj = apprise.Apprise()
        if urls:
            if isinstance(urls, str):
                urls = [urls]
            for url in urls:
                if url.strip():
                    self.apobj.add(url.strip())

    def send(self, title, msg):
        """发送通知"""
        if not self.apobj:
            print("未配置推送服务，不发送推送")
            return
        
        try:
            # 发送通知，支持多种推送方式
            result = self.apobj.notify(
                body=msg,
                title=title
            )
            if result:
                logger.info(f"推送通知发送成功: {title}")
            else:
                logger.warning("推送通知发送失败")
        except Exception as e:
            logger.error(f"推送通知发送异常: {e}")
            import traceback
            logger.error(f"详细错误信息: {traceback.format_exc()}")




def get_url_params(url, key):
    """原始参数值获取"""
    parsed = urlparse(url)

    for param in parsed.query.split("&"):
        if "=" in param:
            k, v = param.split("=", 1)  # 只分割第一个等号
            if k == key:
                return v  # 直接返回原始值

    # 处理片段部分
    if parsed.fragment:
        for param in parsed.fragment.split("&"):
            if "=" in param:
                k, v = param.split("=", 1)
                if k == key:
                    return v

    return None


def days_between_dates(date1_str, date2_str):
    try:
        date1 = datetime.strptime(date1_str, "%Y-%m-%d")
        date2 = datetime.strptime(date2_str, "%Y-%m-%d")
        time_difference = date2 - date1
        days_difference = time_difference.days + 1
        return days_difference
    except Exception as error:
        print("计算两个日期差失败:", error)
        return 0


def get_future_date(date_str, days):
    date_obj = datetime.strptime(date_str, "%Y-%m-%d").date()
    future_date = date_obj + timedelta(days=days)
    return future_date.strftime("%Y-%m-%d")


def _get_or_create_key():
    """获取或创建加密密钥"""
    key_file = "url_key.key"
    if os.path.exists(key_file):
        with open(key_file, "rb") as f:
            return f.read()
    else:
        key = Fernet.generate_key()
        with open(key_file, "wb") as f:
            f.write(key)
        return key


def encrypt_url(url: str) -> str:
    """加密URL"""
    if not url:
        return ""
    try:
        key = _get_or_create_key()
        fernet = Fernet(key)
        encrypted_url = fernet.encrypt(url.encode())
        return base64.b64encode(encrypted_url).decode()
    except Exception as e:
        logger.error(f"URL加密失败: {e}")
        return ""


def decrypt_url(encrypted_url: str) -> str:
    """解密URL"""
    if not encrypted_url:
        return ""
    try:
        key = _get_or_create_key()
        fernet = Fernet(key)
        encrypted_data = base64.b64decode(encrypted_url.encode())
        decrypted_url = fernet.decrypt(encrypted_data)
        return decrypted_url.decode()
    except Exception as e:
        logger.error(f"URL解密失败: {e}")
        return ""