diff options
| -rw-r--r-- | .gitignore | 91 | ||||
| -rw-r--r-- | .gitmodules | 3 | ||||
| -rw-r--r-- | api.py | 124 | ||||
| m--------- | joblib | 0 | ||||
| -rw-r--r-- | requirements.txt | 5 |
5 files changed, 223 insertions, 0 deletions
diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..9a05e2d --- /dev/null +++ b/.gitignore @@ -0,0 +1,91 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +env/ +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*,cover +.hypothesis/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# pyenv +.python-version + +# celery beat schedule file +celerybeat-schedule + +# dotenv +.env + +# virtualenv +.venv/ +venv/ +ENV/ + +# Spyder project settings +.spyderproject + +# Rope project settings +.ropeproject diff --git a/.gitmodules b/.gitmodules new file mode 100644 index 0000000..47695ea --- /dev/null +++ b/.gitmodules @@ -0,0 +1,3 @@ +[submodule "joblib"] + path = joblib + url = https://gitlab.com/vainglorygame/joblib.git @@ -0,0 +1,124 @@ +#!/usr/bin/python3 + +import asyncio +import os +import glob +import logging +import json +import asyncpg + +import joblib.worker + +queue_db = { + "host": os.environ.get("POSTGRESQL_SOURCE_HOST") or "localhost", + "port": os.environ.get("POSTGRESQL_SOURCE_PORT") or 5433, + "user": os.environ.get("POSTGRESQL_SOURCE_USER") or "vainraw", + "password": os.environ.get("POSTGRESQL_SOURCE_PASSWORD") or "vainraw", + "database": os.environ.get("POSTGRESQL_SOURCE_DB") or "vainsocial-raw" +} + +db_config = { + "host": os.environ.get("POSTGRESQL_DEST_HOST") or "localhost", + "port": os.environ.get("POSTGRESQL_DEST_PORT") or 5432, + "user": os.environ.get("POSTGRESQL_DEST_USER") or "vainweb", + "password": os.environ.get("POSTGRESQL_DEST_PASSWORD") or "vainweb", + "database": os.environ.get("POSTGRESQL_DEST_DB") or "vainsocial-web" +} + + +class Cruncher(joblib.worker.Worker): + def __init__(self): + self._con = None + super().__init__(jobtype="crunch") + + async def connect(self, dbconf, queuedb): + """Connect to database.""" + logging.warning("connecting to database") + await super().connect(**queuedb) + self._con = await asyncpg.connect(**dbconf) + + async def setup(self): + pass + + async def _windup(self): + self._tr = self._con.transaction() + await self._tr.start() + + async def _teardown(self, failed): + if failed: + await self._tr.rollback() + else: + await self._tr.commit() + + async def _execute_job(self, jobid, payload, priority): + """Finish a job.""" + dimension_id = payload["dimension"] + + logging.debug("%s: crunching dimension %s", + jobid, dimension_id) + + dimension = await self._con.fetchrow(""" + SELECT * FROM stats_dimensions WHERE id=$1 + """, dimension_id) + if dimension is None: + logging.warning("%s: invalid dimension '%s', exiting", + jobid, dimension_id) + raise joblib.worker.JobFailed("invalid dimension '" + dimension_id + + "'", + False) + + table = dimension["dimension_on"] + field = dimension["name"] + value = dimension["value"] + + if table == "hero": + filter_query = "" + if field == "patch": + filter_query = "WHERE match.patch_version='" + value + "'" + if field == "recent_matches": + filter_query = "WHERE match.api_id IN (SELECT api_id FROM match ORDER BY created_at DESC LIMIT " + value + ")" + + stats = await self._con.fetch(""" + SELECT + heros.id, + SUM(participant.winner::INT)/COUNT(participant.winner)::FLOAT AS win_rate, + SUM(60*participant.minion_kills/match.duration::FLOAT)/COUNT(participant.farm)::FLOAT AS cs_per_min, + 0 AS gold_per_min + FROM participant + + JOIN participant_stats ON participant.api_id=participant_stats.participant_api_id + JOIN roster ON roster.api_id=participant.roster_api_id + JOIN match ON match.api_id=roster.match_api_id + + JOIN heros ON heros.api_name=hero + """ + + filter_query + + """ + GROUP BY heros.id + """) + + for stat in stats: + stat_id = await self._con.fetchrow(""" + INSERT INTO stats(win_rate, cs_per_min, gold_per_min) + VALUES($1, $2, $3) + RETURNING stats.id + """, stat["win_rate"], stat["cs_per_min"], + stat["gold_per_min"]) + await self._con.fetch(""" + INSERT INTO hero_stats(hero_id, dimension_id, stats_id) + VALUES($1, $2, $3) + """, stat["id"], dimension["id"], + stat_id["id"]) + +async def startup(): + worker = Cruncher() + await worker.connect(db_config, queue_db) + await worker.setup() + await worker.start(batchlimit=1) + + +logging.basicConfig(level=logging.DEBUG) + +loop = asyncio.get_event_loop() +loop.run_until_complete(startup()) +loop.run_forever() diff --git a/joblib b/joblib new file mode 160000 +Subproject 7ced841aa44a7dd7bd2d3a68ceb989ef488ce44 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..2e1ea39 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +appdirs==1.4.2 +asyncpg==0.8.4 +packaging==16.8 +pyparsing==2.1.10 +six==1.10.0 |