<gh_stars>0
# updater.py #
import asyncio
import logging
from asyncio import CancelledError
from datetime import timedelta

from asyncpraw import models
from sqlalchemy import select, text, update
from sqlalchemy.exc import PendingRollbackError
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql.functions import now

from models import Submission, Comment
from observer import Observer
from praw_client import PrawClient
from sa_config import Session


class Updater(Observer):
    def __init__(self, **kwargs: dict[str: str]):
        super().__init__()
        self._type = kwargs['type']
        self._Model = Comment if self._type == 'comments' else Submission
        self._interval = kwargs['interval']
        self._threshold = 24 if 'threshold' not in kwargs or kwargs['threshold'] is None else kwargs['threshold']
        self._info_request_size = 1000
        self._task = None
        self._timer = None

    async def _commit(self, session: AsyncSession) -> None:
        try:
            await session.commit()
        except PendingRollbackError:
            await session.rollback()
        except Exception as ex:
            logging.warning(f'{self._type} updater: commit: {ex}')

    async def _session_update(self, session: AsyncSession, model: models) -> None:
        values = {
            'score': int(model.score),
            'score_history': text(f"array_append(score_history, {model.score})")
        }

        if self._type == 'submissions':
            ratio = model.upvote_ratio
            ups = round((ratio * float(model.score)) / (2 * ratio - 1)) if ratio != 0.5 else round(model.score / 2)
            values['ups'] = text(f"array_append(ups, {ups})")
            values['downs'] = text(f"array_append(downs, {ups - model.score})")

        await session.execute(
            update(self._Model)
                .where(self._Model.id == model.id)
                .values(**values))

    async def _run(self) -> None:
        async with Session() as session:
            async with PrawClient() as praw_client:
                try:
                    res = await session.execute(
                        select(self._Model.id)
                            .where(self._Model.updated_at >= now() - timedelta(hours=self._threshold))
                            .filter(self._Model.updated_at <= now() - timedelta(minutes=self._interval))
                    )
                    id_list = [('t1_' if self._type == 'comments' else 't3_') + r[0] for r in res]
                    for i in range((len(id_list) // self._info_request_size) + 1):
                        if self._kill_signal:
                            await self._commit(session)
                            break

                        coro = [self._session_update(session, model) async for model in praw_client.info(
                            id_list[i * self._info_request_size:(i + 1) * self._info_request_size])]

                        await asyncio.gather(*coro, return_exceptions=True)
                        await self._commit(session)
                except Exception as ex:
                    logging.warning(f'{self._type} updater: {ex}')
                finally:
                    self._task = None

    def stop(self) -> None:
        try:
            if self._timer:
                self._timer.cancel()
                self._timer = None
        except CancelledError:
            pass
        finally:
            logging.info(f'■ {self.type} updater')

    async def start(self) -> None:
        if self._type != 'comments' and self._type != 'submissions':
            raise SyntaxWarning("ScoreUpdate.type must be 'comments' or 'submissions'")

        logging.info(f"▶ {self._type} updater")

        while not self._kill_signal:
            try:
                self._task = asyncio.create_task(self._run())
                self._timer = asyncio.create_task(asyncio.sleep(self._interval * 60))
                await self._timer
                while self._task:
                    self._timer = asyncio.create_task(asyncio.sleep(1))
                    await self._timer
                self._timer = None
            except CancelledError:
                return

    def update(self, arg: any) -> None:
        self._observer_state = arg
        if self._kill_signal:
            self.stop()

    @property
    def _kill_signal(self):
        return self._observer_state['kill_signal']

    @property
    def type(self):
        return self._type

    @property
    def interval(self):
        return self._interval

    @property
    def threshold(self):
        return self._threshold

# updater.py #
import asyncio
import logging
from asyncio import CancelledError
from datetime import timedelta

from asyncpraw import models
from sqlalchemy import select, text, update
from sqlalchemy.exc import PendingRollbackError
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql.functions import now

from models import Submission, Comment
from observer import Observer
from praw_client import PrawClient
from sa_config import Session


class Updater(Observer):
    def __init__(self, **kwargs: dict[str: str]):
        super().__init__()
        self._type = kwargs['type']
        self._Model = Comment if self._type == 'comments' else Submission
        self._interval = kwargs['interval']
        self._threshold = 24 if 'threshold' not in kwargs or kwargs['threshold'] is None else kwargs['threshold']
        self._info_request_size = 1000
        self._task = None
        self._timer = None

    async def _commit(self, session: AsyncSession) -> None:
        try:
            await session.commit()
        except PendingRollbackError:
            await session.rollback()
        except Exception as ex:
            logging.warning(f'{self._type} updater: commit: {ex}')

    async def _session_update(self, session: AsyncSession, model: models) -> None:
        values = {
            'score': int(model.score),
            'score_history': text(f"array_append(score_history, {model.score})")
        }

        if self._type == 'submissions':
            ratio = model.upvote_ratio
            ups = round((ratio * float(model.score)) / (2 * ratio - 1)) if ratio != 0.5 else round(model.score / 2)
            values['ups'] = text(f"array_append(ups, {ups})")
            values['downs'] = text(f"array_append(downs, {ups - model.score})")

        await session.execute(
            update(self._Model)
                .where(self._Model.id == model.id)
                .values(**values))

    async def _run(self) -> None:
        async with Session() as session:
            async with PrawClient() as praw_client:
                try:
                    res = await session.execute(
                        select(self._Model.id)
                            .where(self._Model.updated_at >= now() - timedelta(hours=self._threshold))
                            .filter(self._Model.updated_at <= now() - timedelta(minutes=self._interval))
                    )
                    id_list = [('t1_' if self._type == 'comments' else 't3_') + r[0] for r in res]
                    for i in range((len(id_list) // self._info_request_size) + 1):
                        if self._kill_signal:
                            await self._commit(session)
                            break

                        coro = [self._session_update(session, model) async for model in praw_client.info(
                            id_list[i * self._info_request_size:(i + 1) * self._info_request_size])]

                        await asyncio.gather(*coro, return_exceptions=True)
                        await self._commit(session)
                except Exception as ex:
                    logging.warning(f'{self._type} updater: {ex}')
                finally:
                    self._task = None

    def stop(self) -> None:
        try:
            if self._timer:
                self._timer.cancel()
                self._timer = None
        except CancelledError:
            pass
        finally:
            logging.info(f'■ {self.type} updater')

    async def start(self) -> None:
        if self._type != 'comments' and self._type != 'submissions':
            raise SyntaxWarning("ScoreUpdate.type must be 'comments' or 'submissions'")

        logging.info(f"▶ {self._type} updater")

        while not self._kill_signal:
            try:
                self._task = asyncio.create_task(self._run())
                self._timer = asyncio.create_task(asyncio.sleep(self._interval * 60))
                await self._timer
                while self._task:
                    self._timer = asyncio.create_task(asyncio.sleep(1))
                    await self._timer
                self._timer = None
            except CancelledError:
                return

    def update(self, arg: any) -> None:
        self._observer_state = arg
        if self._kill_signal:
            self.stop()

    @property
    def _kill_signal(self):
        return self._observer_state['kill_signal']

    @property
    def type(self):
        return self._type

    @property
    def interval(self):
        return self._interval

    @property
    def threshold(self):
        return self._threshold
# updater.py #