api_rssbot.RssRobot.main() - Code Metrics - Inspection of ":zap: 分批发送：rssbot" - OreosLab/checkinpanel - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 670ff2...8eb876 )

by Leon

created 2022-02-25 16:40 UTC

api_rssbot.RssRobot.main() C

↳ Parent: api_rssbot

Complexity

Conditions

Size

Total Lines	45
Code Lines	32

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	9
eloc	32
nop	1
dl	0
loc	45
rs	6.6666
c	0
b	0
f	0

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

from datetime import datetime, timedelta, timezone

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 = ""
        no = 1
        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:
                # print(entry["published"])
                # print(datetime.now())
                pub_t = datetime.strptime(
                    entry["published"],
                    rss.date_type,
                )

                if pub_t.tzinfo is None:
                    pub_t = pub_t.replace(tzinfo=timezone.utc)
                # print(pub_t)

                if (
                    entry.link not in post_url_list
                    and (datetime.timestamp(datetime.now()) - datetime.timestamp(pub_t))
                    < rss.before * 86400
                ):
                    msg = msg + f"{str(no).zfill(2)}.{entry.title}\n{entry.link}\n\n"
                    if no % 10 == 0:
                        send("RSS 订阅", msg)
                        msg = ""
                    no += 1
                    rss_history_list.append(History(url=entry.link))

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

        if no % 10 != 0:
            send("RSS 订阅", 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()


1			# -- encoding: utf-8 --
2			"""
3			cron: /15 6-22/2 * *
4			new Env('RSS 订阅');
5			"""
6
7			from datetime import datetime, timedelta, timezone
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
19			rss_list = Rss.select()
20			msg = ""
21			no = 1
22			post_url_list = [
23			rss_history.url
24			for rss_history in History.select().where(
25			History.publish_at == datetime.today().strftime("%Y-%m-%d")
26			)
27			]
28
29			for rss in rss_list:
30			rss_history_list = []
31			feed = feedparser.parse(rss.feed)
32			for entry in feed.entries:
33			# print(entry["published"])
34			# print(datetime.now())
35			pub_t = datetime.strptime(
36			entry["published"],
37			rss.date_type,
38			)
39
40			if pub_t.tzinfo is None:
41			pub_t = pub_t.replace(tzinfo=timezone.utc)
42			# print(pub_t)
43
44			if (
45			entry.link not in post_url_list
46			and (datetime.timestamp(datetime.now()) - datetime.timestamp(pub_t))
47			< rss.before * 86400
48			):
49			msg = msg + f"{str(no).zfill(2)}.{entry.title}\n{entry.link}\n\n"
50			if no % 10 == 0:
51			send("RSS 订阅", msg)
52			msg = ""
53			no += 1
54			rss_history_list.append(History(url=entry.link))
55
56			with db.atomic():
57			History.bulk_create(rss_history_list, batch_size=10)
58
59			if no % 10 != 0:
60			send("RSS 订阅", msg)
61
62			def remove_old_history(self):
63			# 只保留最近一周的记录
64			week_date_range = datetime.now() + timedelta(days=-7)
65			History.delete().where(
66			History.publish_at < week_date_range.strftime("%Y-%m-%d")
67			).execute()
68
69
70			if __name__ == "__main__":
71			res = RssRobot().main()
72

OreosLab / checkinpanel

Push — master ( 670ff2...8eb876 )

api_rssbot.RssRobot.main() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like