WIP
This commit is contained in:
Igor Lobanov
@@ -1,207 +0,0 @@
|
||||
import asyncio
|
||||
import time
|
||||
from base.orm import local_session
|
||||
from orm.reaction import ReactionKind, Reaction
|
||||
from services.zine.topics import TopicStorage
|
||||
from services.stat.viewed import ViewedStorage
|
||||
|
||||
|
||||
def kind_to_rate(kind) -> int:
|
||||
if kind in [
|
||||
ReactionKind.AGREE,
|
||||
ReactionKind.LIKE,
|
||||
ReactionKind.PROOF,
|
||||
ReactionKind.ACCEPT,
|
||||
]:
|
||||
return 1
|
||||
elif kind in [
|
||||
ReactionKind.DISAGREE,
|
||||
ReactionKind.DISLIKE,
|
||||
ReactionKind.DISPROOF,
|
||||
ReactionKind.REJECT,
|
||||
]:
|
||||
return -1
|
||||
else:
|
||||
return 0
|
||||
|
||||
|
||||
class ReactedStorage:
|
||||
reacted = {"shouts": {}, "topics": {}, "reactions": {}, "authors": {}}
|
||||
rating = {"shouts": {}, "topics": {}, "reactions": {}}
|
||||
reactions = []
|
||||
to_flush = []
|
||||
period = 30 * 60 # sec
|
||||
lock = asyncio.Lock()
|
||||
modified_shouts = set([])
|
||||
|
||||
@staticmethod
|
||||
async def get_shout_stat(slug, rating):
|
||||
viewed = int(await ViewedStorage.get_shout(slug))
|
||||
# print(viewed)
|
||||
return {
|
||||
"viewed": viewed,
|
||||
"reacted": len(await ReactedStorage.get_shout(slug)),
|
||||
"commented": len(await ReactedStorage.get_comments(slug)),
|
||||
# "rating": await ReactedStorage.get_rating(slug),
|
||||
"rating": rating
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
async def get_shout(shout_slug):
|
||||
self = ReactedStorage
|
||||
async with self.lock:
|
||||
return self.reacted["shouts"].get(shout_slug, [])
|
||||
|
||||
@staticmethod
|
||||
async def get_author(user_slug):
|
||||
self = ReactedStorage
|
||||
async with self.lock:
|
||||
return self.reacted["authors"].get(user_slug, [])
|
||||
|
||||
@staticmethod
|
||||
async def get_shouts_by_author(user_slug):
|
||||
self = ReactedStorage
|
||||
async with self.lock:
|
||||
author_reactions = self.reacted["authors"].get(user_slug, [])
|
||||
shouts = []
|
||||
for r in author_reactions:
|
||||
if r.shout not in shouts:
|
||||
shouts.append(r.shout)
|
||||
return shouts
|
||||
|
||||
@staticmethod
|
||||
async def get_topic(topic_slug):
|
||||
self = ReactedStorage
|
||||
async with self.lock:
|
||||
return self.reacted["topics"].get(topic_slug, [])
|
||||
|
||||
@staticmethod
|
||||
async def get_comments(shout_slug):
|
||||
self = ReactedStorage
|
||||
async with self.lock:
|
||||
return list(
|
||||
filter(lambda r: bool(r.body), self.reacted["shouts"].get(shout_slug, {}))
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
async def get_topic_comments(topic_slug):
|
||||
self = ReactedStorage
|
||||
async with self.lock:
|
||||
return list(
|
||||
filter(lambda r: bool(r.body), self.reacted["topics"].get(topic_slug, []))
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
async def get_reaction_comments(reaction_id):
|
||||
self = ReactedStorage
|
||||
async with self.lock:
|
||||
return list(
|
||||
filter(
|
||||
lambda r: bool(r.body), self.reacted["reactions"].get(reaction_id, {})
|
||||
)
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
async def get_reaction(reaction_id):
|
||||
self = ReactedStorage
|
||||
async with self.lock:
|
||||
return self.reacted["reactions"].get(reaction_id, [])
|
||||
|
||||
@staticmethod
|
||||
async def get_rating(shout_slug):
|
||||
self = ReactedStorage
|
||||
rating = 0
|
||||
async with self.lock:
|
||||
for r in self.reacted["shouts"].get(shout_slug, []):
|
||||
rating = rating + kind_to_rate(r.kind)
|
||||
return rating
|
||||
|
||||
@staticmethod
|
||||
async def get_topic_rating(topic_slug):
|
||||
self = ReactedStorage
|
||||
rating = 0
|
||||
async with self.lock:
|
||||
for r in self.reacted["topics"].get(topic_slug, []):
|
||||
rating = rating + kind_to_rate(r.kind)
|
||||
return rating
|
||||
|
||||
@staticmethod
|
||||
async def get_reaction_rating(reaction_id):
|
||||
self = ReactedStorage
|
||||
rating = 0
|
||||
async with self.lock:
|
||||
for r in self.reacted["reactions"].get(reaction_id, []):
|
||||
rating = rating + kind_to_rate(r.kind)
|
||||
return rating
|
||||
|
||||
@staticmethod
|
||||
async def react(reaction):
|
||||
ReactedStorage.modified_shouts.add(reaction.shout)
|
||||
|
||||
@staticmethod
|
||||
async def recount(reactions):
|
||||
self = ReactedStorage
|
||||
for r in reactions:
|
||||
# renew reactions by shout
|
||||
self.reacted["shouts"][r.shout] = self.reacted["shouts"].get(r.shout, [])
|
||||
self.reacted["shouts"][r.shout].append(r)
|
||||
# renew reactions by author
|
||||
self.reacted["authors"][r.createdBy] = self.reacted["authors"].get(r.createdBy, [])
|
||||
self.reacted["authors"][r.createdBy].append(r)
|
||||
# renew reactions by topic
|
||||
shout_topics = await TopicStorage.get_topics_by_slugs([r.shout, ])
|
||||
for t in shout_topics:
|
||||
self.reacted["topics"][t] = self.reacted["topics"].get(t, [])
|
||||
self.reacted["topics"][t].append(r)
|
||||
self.rating["topics"][t] = \
|
||||
self.rating["topics"].get(t, 0) + kind_to_rate(r.kind)
|
||||
if r.replyTo:
|
||||
# renew reactions replies
|
||||
self.reacted["reactions"][r.replyTo] = \
|
||||
self.reacted["reactions"].get(r.replyTo, [])
|
||||
self.reacted["reactions"][r.replyTo].append(r)
|
||||
self.rating["reactions"][r.replyTo] = \
|
||||
self.rating["reactions"].get(r.replyTo, 0) + kind_to_rate(r.kind)
|
||||
else:
|
||||
# renew shout rating
|
||||
self.rating["shouts"][r.shout] = \
|
||||
self.rating["shouts"].get(r.shout, 0) + kind_to_rate(r.kind)
|
||||
|
||||
@staticmethod
|
||||
def init(session):
|
||||
self = ReactedStorage
|
||||
all_reactions = session.query(Reaction).all()
|
||||
self.modified_shouts = list(set([r.shout for r in all_reactions]))
|
||||
print("[stat.reacted] %d shouts with reactions" % len(self.modified_shouts))
|
||||
|
||||
@staticmethod
|
||||
async def recount_changed(session):
|
||||
start = time.time()
|
||||
self = ReactedStorage
|
||||
async with self.lock:
|
||||
sss = list(self.modified_shouts)
|
||||
c = 0
|
||||
for slug in sss:
|
||||
siblings = session.query(Reaction).where(Reaction.shout == slug).all()
|
||||
c += len(siblings)
|
||||
await self.recount(siblings)
|
||||
|
||||
print("[stat.reacted] %d reactions recounted" % c)
|
||||
print("[stat.reacted] %d shouts modified" % len(self.modified_shouts))
|
||||
print("[stat.reacted] %d topics" % len(self.reacted["topics"].values()))
|
||||
print("[stat.reacted] %d authors" % len(self.reacted["authors"].values()))
|
||||
print("[stat.reacted] %d replies" % len(self.reacted["reactions"]))
|
||||
self.modified_shouts = set([])
|
||||
|
||||
end = time.time()
|
||||
print("[stat.reacted] recount_changed took %fs " % (end - start))
|
||||
|
||||
@staticmethod
|
||||
async def worker():
|
||||
while True:
|
||||
try:
|
||||
with local_session() as session:
|
||||
await ReactedStorage.recount_changed(session)
|
||||
except Exception as err:
|
||||
print("[stat.reacted] recount error %s" % (err))
|
||||
await asyncio.sleep(ReactedStorage.period)
|
||||
@@ -1,73 +0,0 @@
|
||||
import asyncio
|
||||
import time
|
||||
from base.orm import local_session
|
||||
from orm.shout import Shout, ShoutTopic, ShoutAuthor
|
||||
from orm.topic import TopicFollower
|
||||
from sqlalchemy.sql.expression import select
|
||||
|
||||
|
||||
class TopicStat:
|
||||
# by slugs
|
||||
shouts_by_topic = {} # Shout object stored
|
||||
authors_by_topic = {} # User
|
||||
followers_by_topic = {} # User
|
||||
#
|
||||
lock = asyncio.Lock()
|
||||
period = 30 * 60 # sec
|
||||
|
||||
@staticmethod
|
||||
async def load_stat(session):
|
||||
start = time.time()
|
||||
self = TopicStat
|
||||
shout_topics = session.query(ShoutTopic, Shout).join(Shout).all()
|
||||
all_shout_authors = session.query(ShoutAuthor).all()
|
||||
print("[stat.topics] %d links for shouts" % len(shout_topics))
|
||||
for [shout_topic, shout] in shout_topics:
|
||||
tpc = shout_topic.topic
|
||||
# shouts by topics
|
||||
# shout = session.query(Shout).where(Shout.slug == shout_topic.shout).first()
|
||||
self.shouts_by_topic[tpc] = self.shouts_by_topic.get(tpc, dict())
|
||||
self.shouts_by_topic[tpc][shout.slug] = shout
|
||||
|
||||
# authors by topics
|
||||
shout_authors = filter(lambda asa: asa.shout == shout.slug, all_shout_authors)
|
||||
|
||||
self.authors_by_topic[tpc] = self.authors_by_topic.get(tpc, dict())
|
||||
for sa in shout_authors:
|
||||
self.authors_by_topic[tpc][sa.shout] = sa.caption
|
||||
|
||||
self.followers_by_topic = {}
|
||||
followings = session.query(TopicFollower).all()
|
||||
print("[stat.topics] %d followings by users" % len(followings))
|
||||
for flw in followings:
|
||||
topic = flw.topic
|
||||
userslug = flw.follower
|
||||
self.followers_by_topic[topic] = self.followers_by_topic.get(topic, dict())
|
||||
self.followers_by_topic[topic][userslug] = userslug
|
||||
|
||||
end = time.time()
|
||||
print("[stat.topics] load_stat took %fs " % (end - start))
|
||||
|
||||
@staticmethod
|
||||
async def get_shouts(topic):
|
||||
self = TopicStat
|
||||
async with self.lock:
|
||||
return self.shouts_by_topic.get(topic, dict())
|
||||
|
||||
@staticmethod
|
||||
async def worker():
|
||||
self = TopicStat
|
||||
first_run = True
|
||||
while True:
|
||||
try:
|
||||
with local_session() as session:
|
||||
async with self.lock:
|
||||
await self.load_stat(session)
|
||||
except Exception as err:
|
||||
raise Exception(err)
|
||||
if first_run:
|
||||
# sleep for period + 1 min after first run
|
||||
# to distribute load on server by workers with the same period
|
||||
await asyncio.sleep(60)
|
||||
first_run = False
|
||||
await asyncio.sleep(self.period)
|
||||
Reference in New Issue
Block a user