nochecks

services/stat/viewed.py

@@ -2,9 +2,10 @@ import asyncio
 
 from gql import Client, gql
 from gql.transport.aiohttp import AIOHTTPTransport
-
 from base.orm import local_session
+from sqlalchemy import func, select
 from orm.viewed import ViewedEntry
+from orm.shout import ShoutTopic
 from services.zine.topics import TopicStorage
 from ssl import create_default_context
 
@@ -43,12 +44,13 @@ ssl = create_default_context()
 class ViewedStorage:
     lock = asyncio.Lock()
     by_topics = {}
+    by_shouts = {}
     period = 5 * 60  # 5 minutes
     client = None
     transport = None
 
     @staticmethod
-    async def load_views(session):
+    async def update_views(session):
         # TODO: when the struture of payload will be transparent
         # TODO: perhaps ackee token getting here
 
@@ -61,18 +63,53 @@ class ViewedStorage:
             print(domains)
             print('\n\n# TODO: something here...\n\n')
 
+    @staticmethod
+    async def get_shout(shout_slug):
+        self = ViewedStorage
+        async with self.lock:
+            r = self.by_shouts.get(shout_slug)
+            if r:
+                with local_session() as session:
+                    shout_views = 0
+                    shout_views_q = select(func.sum(ViewedEntry.amount)).where(
+                        ViewedEntry.shout == shout_slug
+                    )
+                    shout_views = session.execute(shout_views_q)
+                    self.by_shouts[shout_slug] = shout_views
+                    return shout_views
+            else:
+                return r
+
+    @staticmethod
+    async def get_topic(topic_slug):
+        self = ViewedStorage
+        topic_views = 0
+        async with self.lock:
+            topic_views_by_shouts = self.by_topics.get(topic_slug) or {}
+            if len(topic_views_by_shouts.keys()) == 0:
+                with local_session() as session:
+                    shoutslugs = session.query(ShoutTopic.shout).where(ShoutTopic.topic == topic_slug).all()
+                    self.by_topics[topic_slug] = {}
+                    for slug in shoutslugs:
+                        self.by_topics[topic_slug][slug] = await self.get_shout(slug)
+                topic_views_by_shouts = self.by_topics.get(topic_slug) or {}
+            for shout in topic_views_by_shouts:
+                topic_views += shout
+        return topic_views
+
     @staticmethod
     async def increment(shout_slug, amount=1, viewer='anonymous'):
         self = ViewedStorage
         async with self.lock:
             with local_session() as session:
-                viewed = ViewedEntry.create({
+                viewed = ViewedEntry.create(**{
                     "viewer": viewer,
-                    "shout": shout_slug
+                    "shout": shout_slug,
+                    "amount": amount
                 })
                 session.add(viewed)
                 session.commit()
-
+                self.by_shouts[shout_slug] = self.by_shouts.get(shout_slug, 0) + amount
                 shout_topics = await TopicStorage.get_topics_by_slugs([shout_slug, ])
                 for t in shout_topics:
                     self.by_topics[t] = self.by_topics.get(t) or {}
@@ -85,7 +122,7 @@ class ViewedStorage:
         while True:
             try:
                 with local_session() as session:
-                    await self.load_views(session)
+                    await self.update_views(session)
             except Exception as err:
                 print("[stat.viewed] : %s" % (err))
             print("[stat.viewed] renew period: %d minutes" % (self.period / 60))

services/stat/views.py

@@ -1,127 +0,0 @@
-import asyncio
-import json
-
-from gql import Client, gql
-from gql.transport.aiohttp import AIOHTTPTransport
-
-from base.redis import redis
-from services.zine.topics import TopicStorage
-from ssl import create_default_context
-
-
-query_ackee_views = gql(
-    """
-    query getDomainsFacts {
-        domains {
-            statistics {
-                views {
-                    id
-                    count
-                }
-                pages {
-                    id
-                    count
-                    created
-                }
-            }
-            facts {
-                activeVisitors
-                # averageViews
-                # averageDuration
-                viewsToday
-                viewsMonth
-                viewsYear
-            }
-        }
-    }
-    """
-)
-
-ssl = create_default_context()
-
-
-class ViewStat:
-    lock = asyncio.Lock()
-    by_slugs = {}
-    by_topics = {}
-    period = 5 * 60  # 5 minutes
-    transport = AIOHTTPTransport(url="https://ackee.discours.io/", ssl=ssl)
-    client = Client(transport=transport, fetch_schema_from_transport=True)
-
-    @staticmethod
-    async def load_views():
-        # TODO: when the struture of paylod will be transparent
-        # TODO: perhaps ackee token getting here
-
-        self = ViewStat
-        async with self.lock:
-            self.by_topics = await redis.execute("GET", "views_by_topics")
-            if self.by_topics:
-                self.by_topics = dict(json.loads(self.by_topics))
-            else:
-                self.by_topics = {}
-            self.by_slugs = await redis.execute("GET", "views_by_shouts")
-            if self.by_slugs:
-                self.by_slugs = dict(json.loads(self.by_slugs))
-            else:
-                self.by_slugs = {}
-            domains = await self.client.execute_async(query_ackee_views)
-            print("[stat.ackee] loaded domains")
-            print(domains)
-
-            print('\n\n# TODO: something here...\n\n')
-
-    @staticmethod
-    async def get_shout(shout_slug):
-        self = ViewStat
-        async with self.lock:
-            return self.by_slugs.get(shout_slug) or 0
-
-    @staticmethod
-    async def get_topic(topic_slug):
-        self = ViewStat
-        async with self.lock:
-            shouts = self.by_topics.get(topic_slug) or {}
-            topic_views = 0
-            for v in shouts.values():
-                topic_views += v
-            return topic_views
-
-    @staticmethod
-    async def increment(shout_slug, amount=1):
-        self = ViewStat
-        async with self.lock:
-            self.by_slugs[shout_slug] = self.by_slugs.get(shout_slug) or 0
-            self.by_slugs[shout_slug] += amount
-            await redis.execute(
-                "SET",
-                f"views_by_shouts/{shout_slug}",
-                str(self.by_slugs[shout_slug])
-            )
-            shout_topics = await TopicStorage.get_topics_by_slugs([shout_slug, ])
-            for t in shout_topics:
-                self.by_topics[t] = self.by_topics.get(t) or {}
-                self.by_topics[t][shout_slug] = self.by_topics[t].get(shout_slug) or 0
-                self.by_topics[t][shout_slug] += amount
-                await redis.execute(
-                    "SET",
-                    f"views_by_topics/{t}/{shout_slug}",
-                    str(self.by_topics[t][shout_slug])
-                )
-
-    @staticmethod
-    async def reset():
-        self = ViewStat
-        self.by_topics = {}
-        self.by_slugs = {}
-
-    @staticmethod
-    async def worker():
-        self = ViewStat
-        while True:
-            try:
-                await self.load_views()
-            except Exception as err:
-                print("[stat.ackee] : %s" % (err))
-            print("[stat.ackee] renew period: %d minutes" % (ViewStat.period / 60))
-            await asyncio.sleep(self.period)