views from ackee

services/stat/reacted.py

@@ -2,6 +2,7 @@ import asyncio
 from base.orm import local_session
 from orm.reaction import ReactionKind, Reaction
 from services.zine.topics import TopicStorage
+from services.stat.views import Stat
 
 
 def kind_to_rate(kind) -> int:
@@ -35,8 +36,7 @@ class ReactedStorage:
     @staticmethod
     async def get_shout_stat(slug):
         return {
-            # TODO: use ackee as datasource
-            "viewed": 0,  # await ViewedStorage.get_shout(slug),
+            "viewed": await Stat.get_shout(slug),
             "reacted": len(await ReactedStorage.get_shout(slug)),
             "commented": len(await ReactedStorage.get_comments(slug)),
             "rating": await ReactedStorage.get_rating(slug),

services/stat/viewed.py

@@ -1,110 +0,0 @@
-import asyncio
-from datetime import datetime
-
-from base.orm import local_session
-
-from sqlalchemy.orm.attributes import flag_modified
-
-from orm.shout import ShoutTopic
-from orm.viewed import ViewedByDay
-
-
-class ViewedStorage:
-    viewed = {"shouts": {}, "topics": {}, "reactions": {}}
-    this_day_views = {}
-    to_flush = []
-    period = 30 * 60  # sec
-    lock = asyncio.Lock()
-
-    @staticmethod
-    def init(session):
-        self = ViewedStorage
-        views = session.query(ViewedByDay).all()
-
-        for view in views:
-            shout = view.shout
-            topics = (
-                session.query(ShoutTopic.topic).filter(ShoutTopic.shout == shout).all()
-            )
-            value = view.value
-            if shout:
-                old_value = self.viewed["shouts"].get(shout, 0)
-                self.viewed["shouts"][shout] = old_value + value
-            for t in topics:
-                old_topic_value = self.viewed["topics"].get(t, 0)
-                self.viewed["topics"][t] = old_topic_value + value
-            if shout not in self.this_day_views:
-                self.this_day_views[shout] = view
-            this_day_view = self.this_day_views[shout]
-            if this_day_view.day < view.day:
-                self.this_day_views[shout] = view
-
-        print("[stat.viewed] %d shouts viewed" % len(self.viewed['shouts']))
-
-    @staticmethod
-    async def get_shout(shout_slug):
-        self = ViewedStorage
-        async with self.lock:
-            return self.viewed["shouts"].get(shout_slug, 0)
-
-    @staticmethod
-    async def get_topic(topic_slug):
-        self = ViewedStorage
-        async with self.lock:
-            return self.viewed["topics"].get(topic_slug, 0)
-
-    @staticmethod
-    async def get_reaction(reaction_id):
-        self = ViewedStorage
-        async with self.lock:
-            return self.viewed["reactions"].get(reaction_id, 0)
-
-    @staticmethod
-    async def increment(shout_slug, amount=1):
-        self = ViewedStorage
-        async with self.lock:
-            this_day_view = self.this_day_views.get(shout_slug)
-            day_start = datetime.now().replace(hour=0, minute=0, second=0)
-            if not this_day_view or this_day_view.day < day_start:
-                if this_day_view and getattr(this_day_view, "modified", False):
-                    self.to_flush.append(this_day_view)
-                this_day_view = ViewedByDay.create(shout=shout_slug, value=1)
-                self.this_day_views[shout_slug] = this_day_view
-            else:
-                this_day_view.value = this_day_view.value + amount
-            this_day_view.modified = True
-            self.viewed["shouts"][shout_slug] = (self.viewed["shouts"].get(shout_slug, 0) + amount)
-            with local_session() as session:
-                topics = (
-                    session.query(ShoutTopic.topic)
-                    .where(ShoutTopic.shout == shout_slug)
-                    .all()
-                )
-                for t in topics:
-                    self.viewed["topics"][t] = self.viewed["topics"].get(t, 0) + amount
-            flag_modified(this_day_view, "value")
-
-    @staticmethod
-    async def flush_changes(session):
-        self = ViewedStorage
-        async with self.lock:
-            for view in self.this_day_views.values():
-                if getattr(view, "modified", False):
-                    session.add(view)
-                    flag_modified(view, "value")
-                    view.modified = False
-            for view in self.to_flush:
-                session.add(view)
-            self.to_flush.clear()
-        session.commit()
-
-    @staticmethod
-    async def worker():
-        while True:
-            try:
-                with local_session() as session:
-                    await ViewedStorage.flush_changes(session)
-                    print("[stat.viewed] periodical flush")
-            except Exception as err:
-                print("[stat.viewed] : %s" % (err))
-            await asyncio.sleep(ViewedStorage.period)

services/stat/views.py

@@ -0,0 +1,99 @@
+from gql import gql, Client
+from gql.transport.aiohttp import AIOHTTPTransport
+import asyncio
+
+from services.zine.topics import TopicStorage
+
+query_ackee_views = gql(
+    """
+    query getDomainsFacts {
+        domains {
+            statistics {
+                views {
+                    id
+                    count
+                }
+                pages {
+                    id
+                    count
+                    created
+                }
+            }
+            facts {
+                activeVisitors
+                # averageViews
+                # averageDuration
+                viewsToday
+                viewsMonth
+                viewsYear
+            }
+        }
+    }
+    """
+)
+
+
+class Stat:
+    lock = asyncio.Lock()
+    by_slugs = {}
+    by_topics = {}
+    period = 30 * 60  # 30 minutes
+    transport = AIOHTTPTransport(url="https://ackee.discours.io/")
+    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 = Stat
+        async with self.lock:
+            domains = self.client.execute(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 = Stat
+        async with self.lock:
+            return self.by_slugs.get(shout_slug) or 0
+
+    @staticmethod
+    async def get_topic(topic_slug):
+        self = Stat
+        async with self.lock:
+            shouts = self.by_topics.get(topic_slug)
+            topic_views = 0
+            for v in shouts.values():
+                topic_views += v
+            return topic_views
+
+    @staticmethod
+    async def increment(shout_slug, amount=1):
+        self = Stat
+        async with self.lock:
+            self.by_slugs[shout_slug] = self.by_slugs.get(shout_slug) or 0
+            self.by_slugs[shout_slug] += 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 {}
+                self.by_topics[t][shout_slug] = self.by_topics[t].get(shout_slug) or 0
+                self.by_topics[t][shout_slug] += amount
+
+    @staticmethod
+    async def update():
+        self = Stat
+        async with self.lock:
+            self.load_views()
+
+    @staticmethod
+    async def worker():
+        while True:
+            try:
+                await Stat.update()
+            except Exception as err:
+                print("[stat.ackee] : %s" % (err))
+            print("[stat.ackee] renew period: %d minutes" % (Stat.period / 60))
+            await asyncio.sleep(Stat.period)