rssbot.RssRobot.main() - Code Metrics - Inspection of "add rssbot" - OreosLab/checkinpanel - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#90)

unknown

created 2022-01-22 01:43 UTC

rssbot.RssRobot.main() A

↳ Parent: rssbot

Complexity

Conditions

Size

Total Lines	18
Code Lines	16

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	5
eloc	16
nop	1
dl	0
loc	18
rs	9.1333
c	0
b	0
f	0

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

import feedparser

from datetime import datetime, timedelta
from models import db, Rss, History
from notify_mtr import send


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

OreosLab / checkinpanel

Pull Request — master (#90)

rssbot.RssRobot.main() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like