api_rssbot.RssRobot.main() - Code Metrics - Inspection of ":sparkles: 恢复更新" - OreosLab/checkinpanel - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( d5e461...9fd0c7 )

by Leon

created 2022-01-25 14:30 UTC

api_rssbot.RssRobot.main() A

↳ Parent: api_rssbot

Complexity

Conditions

Size

Total Lines	22
Code Lines	18

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	5
eloc	18
nop	1
dl	0
loc	22
rs	9.0333
c	0
b	0
f	0

# -*- encoding: utf-8 -*-
"""
cron: */15 6-22 * * *
new Env('RSS 订阅推送');
"""

from datetime import datetime, timedelta

import feedparser

from notify_mtr import send
from utils_models import History, Rss, db


class RssRobot:
    def main(self):
        self.remove_old_history()
        rss_list = Rss.select()
        msg = ""
        post_url_list = [
            rss_history.url
            for rss_history in History.select().where(
                History.publish_at == datetime.today().strftime("%Y-%m-%d")
            )
        ]
        for rss in rss_list:
            rss_history_list = []
            feed = feedparser.parse(rss.feed)
            for entry in feed.entries:
                if entry.link not in post_url_list:
                    msg = msg + f"{entry.title}\n{entry.link}\n\n"
                    rss_history_list.append(History(url=entry.link))

            with db.atomic():
                History.bulk_create(rss_history_list, batch_size=10)

        return msg

    def remove_old_history(self):
        # 只保留最近一周的记录
        week_date_range = datetime.now() + timedelta(days=-7)
        History.delete().where(
            History.publish_at < week_date_range.strftime("%Y-%m-%d")
        ).execute()


if __name__ == "__main__":
    res = RssRobot().main()
    send("RSS 订阅", res)


1			# -- encoding: utf-8 --
2			"""
3			cron: /15 6-22 * *
4			new Env('RSS 订阅推送');
5			"""
6
7			from datetime import datetime, timedelta
8
9			import feedparser
10
11			from notify_mtr import send
12			from utils_models import History, Rss, db
13
14
15			class RssRobot:
16			def main(self):
17			self.remove_old_history()
18			rss_list = Rss.select()
19			msg = ""
20			post_url_list = [
21			rss_history.url
22			for rss_history in History.select().where(
23			History.publish_at == datetime.today().strftime("%Y-%m-%d")
24			)
25			]
26			for rss in rss_list:
27			rss_history_list = []
28			feed = feedparser.parse(rss.feed)
29			for entry in feed.entries:
30			if entry.link not in post_url_list:
31			msg = msg + f"{entry.title}\n{entry.link}\n\n"
32			rss_history_list.append(History(url=entry.link))
33
34			with db.atomic():
35			History.bulk_create(rss_history_list, batch_size=10)
36
37			return msg
38
39			def remove_old_history(self):
40			# 只保留最近一周的记录
41			week_date_range = datetime.now() + timedelta(days=-7)
42			History.delete().where(
43			History.publish_at < week_date_range.strftime("%Y-%m-%d")
44			).execute()
45
46
47			if __name__ == "__main__":
48			res = RssRobot().main()
49			send("RSS 订阅", res)
50

OreosLab / checkinpanel

Push — master ( d5e461...9fd0c7 )

api_rssbot.RssRobot.main() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like