refactor

- fix rename bug
- temp save

src/module/network/__init__.py

@@ -0,0 +1,2 @@
+from .request_contents import RequestContent
+from .notification import PostNotification, ServerChanNotification

src/module/network/notification.py

@@ -0,0 +1,62 @@
+import logging
+
+import requests
+
+from module.conf import settings
+
+
+logger = logging.getLogger(__name__)
+
+
+class PostNotification:
+    def __init__(self):
+        self.token = settings.notification_token
+        self.notification_url = lambda message: f"https://api.pushbullet.com/v2/{self.token}/{message}"
+
+    def ifttt_post(self, message):
+        url = self.notification_url(message)
+        response = requests.get(url)
+        return response.status_code == 200
+
+
+class TelegramNotification:
+    def __init__(self):
+        self.token = settings.notification_token
+        self.notification_url = f"https://api.telegram.org/bot{self.token}/sendMessage"
+
+    def send_msg(self, title: str, desp: str) -> bool:
+        if not settings.notification_enable:
+            return False
+        data = {
+            "chat_id": settings.notification_chat_id,
+            "text": f"{title}\n{desp}",
+        }
+
+
+class ServerChanNotification:
+    """Server酱推送"""
+
+    def __init__(self):
+        self.token = settings.notification.token
+        self.notification_url = f"https://sctapi.ftqq.com/{self.token}.send"
+
+    def send_msg(self, title: str, desp: str) -> bool:
+        if not settings.notification.enable:
+            return False
+        data = {
+            "title": title,
+            "desp": desp,
+        }
+        try:
+            resp = requests.post(self.notification_url, json=data, timeout=3)
+            resp.raise_for_status()
+        except requests.RequestException as e:
+            logging.error("[ServerChanNotification] send fail, error: %s" % e)
+            return False
+        return True
+
+
+if __name__ == '__main__':
+    name = "勇者、辞职不干了"
+    notification = ServerChanNotification()
+    notification.send_msg(f"《{name[:10]}》缓存成功", f"[Auto Bangumi]《{name}》缓存成功")

src/module/network/request_contents.py

@@ -0,0 +1,43 @@
+from dataclasses import dataclass
+
+from bs4 import BeautifulSoup
+
+from .request_url import RequestURL
+
+from module.conf import settings
+
+import re
+
+FILTER = "|".join(settings.rss_parser.filter)
+
+
+@dataclass
+class TorrentInfo:
+    name: str
+    torrent_link: str
+
+
+class RequestContent(RequestURL):
+    # Mikanani RSS
+    def get_torrents(self, _url: str) -> [TorrentInfo]:
+        soup = self.get_xml(_url)
+        torrent_titles = [item.title.string for item in soup.find_all("item")]
+        torrent_urls = [item.get("url") for item in soup.find_all("enclosure")]
+        torrents = []
+        for _title, torrent_url in zip(torrent_titles, torrent_urls):
+            if re.search(FILTER, _title) is None:
+                torrents.append(TorrentInfo(_title, torrent_url))
+        return torrents
+
+    def get_torrent(self, _url) -> TorrentInfo:
+        soup = self.get_xml(_url)
+        item = soup.find("item")
+        enclosure = item.find("enclosure")
+        return TorrentInfo(item.title.string, enclosure["url"])
+
+    def get_xml(self, url):
+        return BeautifulSoup(self.get_url(url).text, "xml")
+
+    # API JSON
+    def get_json(self, _url) -> dict:
+        return self.get_url(_url).json()

src/module/network/request_url.py

@@ -0,0 +1,55 @@
+import time
+
+import requests
+import socket
+import socks
+import logging
+
+from module.conf import settings
+
+logger = logging.getLogger(__name__)
+
+
+class RequestURL:
+    def __init__(self):
+        self.header = {
+            "user-agent": "Mozilla/5.0",
+            "Accept": "application/xml"
+        }
+
+    def get_url(self, url):
+        times = 0
+        while times < 5:
+            try:
+                req = self.session.get(url=url, headers=self.header)
+                req.raise_for_status()
+                return req
+            except requests.RequestException as e:
+                logger.debug(f"URL: {url}")
+                logger.debug(e)
+                logger.warning("ERROR with Connection.Please check DNS/Connection settings")
+                time.sleep(5)
+                times += 1
+            except Exception as e:
+                logger.debug(f"URL: {url}")
+                logger.debug(e)
+                break
+
+    def __enter__(self):
+        self.session = requests.Session()
+        if settings.proxy.enable:
+            if settings.proxy.type == "http":
+                url = f"http://{settings.proxy.host}:{settings.proxy.port}"
+                self.session.proxies = {
+                    "https": url,
+                    "http": url,
+                }
+            elif settings.proxy.type == "socks5":
+                socks.set_default_proxy(socks.SOCKS5, addr=settings.proxy.host, port=settings.proxy.port, rdns=True,
+                                        username=settings.proxy.username, password=settings.proxy.password)
+                socket.socket = socks.socksocket
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.session.close()
+
+