init

.gitignore

@@ -0,0 +1,106 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# 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/
+.pytest_cache/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+
+# 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
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+
+.vscode/

README.md

@@ -1,2 +1,13 @@
 # overwatch_stats
-Collecting statistics for overwatch players
+
+Environments:
+Basically the project runs on Python3, for detailed dependencies, see requirements.txt
+
+for running the script `get_token.py`, one should install `browser-cookie3` and `requests` first, by running
+
+```
+pip install browser-cookie3
+pip install requests
+```
+
+in terminal.

app/__init__.py

@@ -0,0 +1,3 @@
+import logging
+
+logging.basicConfig(level=logging.INFO)

app/bnet_retriever/_plugin.py

@@ -0,0 +1,7 @@
+from bson import ObjectId
+
+
+class Plugin:
+    def __init__(self, runner: 'Runner'):
+        self.root: 'Runner' = runner
+        self._id: ObjectId = runner._id

app/bnet_retriever/api.py

@@ -0,0 +1,37 @@
+import requests
+from functools import wraps
+import logging
+
+logger = logging.getLogger(__name__)
+
+GAMEDATA_URL = 'https://ow.blizzard.cn/action/career/profile/gamedata'
+CAREER_URL = 'https://ow.blizzard.cn/action/career/profile'
+
+
+def bnet_api(func):
+    @wraps(func)
+    def decorated(*args, **kwargs):
+        resp = func(*args, **kwargs)
+        if resp.status_code != 200:
+            logger.info('call [%s] failed, 200, args:[%s] [%s]', func.__name__, args, kwargs)
+            return False
+        json = resp.json()
+        if json.get('status'):
+            if json['status'] != 'success':
+                logger.info('call [%s] failed, status non-success , args:[%s] [%s]', func.__name__, args, kwargs)
+                return False
+            return json['data']
+        return json
+    return decorated
+
+
+@bnet_api
+def get_gamedata():
+    return requests.get(GAMEDATA_URL)
+
+
+@bnet_api
+def get_profile(cred):
+    return requests.get(CAREER_URL, cookies={
+        'bnet_user_cred': cred
+    })

app/bnet_retriever/credential.py

@@ -0,0 +1,42 @@
+import json
+import os
+import time
+from typing import Dict, Optional
+
+from bson import ObjectId
+
+from app.bnet_retriever._plugin import Plugin
+from app.bnet_retriever.utils import serialize
+from app.bnet_retriever.utils.login import login
+from app.utils.db import Mongo
+
+
+class CredentialManager(Plugin):
+    def __init__(self, runner):
+        super(CredentialManager, self).__init__(runner)
+        self._bnet_user_cred: Optional[Dict] = None
+
+    @property
+    def raw_cred(self) -> Dict:
+        return serialize.fromstr(Mongo.db.user.find_one({'_id': self._id})['credential'])
+
+    @property
+    def bnet_user_cred(self) -> str:
+        # Here, self.last_updated_at records if the bnet_user_cred outdated ( 1 hr expiration)
+        # Different from the lastUpdatedAt field in mongodb
+        if self._bnet_user_cred is None or time.time() - self.last_updated_at > 60*55:
+            bnet_user_cred, new_cred = login(self.raw_cred)
+            self._bnet_user_cred = bnet_user_cred
+            self.update_raw_cred(new_cred)
+            self.last_updated_at = time.time()
+        return self._bnet_user_cred
+
+    def update_raw_cred(self, new_cred: Dict) -> None:
+        Mongo.db.user.update_one({'_id': self._id}, {
+            '$set': {
+                'credential': serialize.tostr(new_cred),
+            },
+            '$currentDate': {
+                'lastUpdatedAt': True
+            }
+        })

app/bnet_retriever/profile.py

@@ -0,0 +1,85 @@
+from app.bnet_retriever._plugin import Plugin
+from app.bnet_retriever import api
+from app.bnet_retriever.utils import reformat
+from app.utils.db import Mongo
+from typing import Dict
+from bson import ObjectId
+import logging
+import pymongo
+
+gamedata = api.get_gamedata()
+
+logger = logging.getLogger(__name__)
+
+
+class Profile(Plugin):
+    def __init__(self, runner, func):
+        super(Profile, self).__init__(runner)
+        self._func = func
+        # Lazy
+        self._raw = None
+        self._formatted = None
+
+    @property
+    def raw(self):
+        if self._raw is None:
+            self._raw = self._func(self)
+        return self._raw
+
+    @property
+    def formatted(self):
+        if self._formatted is None:
+            self._formatted = {
+                'career': reformat.detailed_mapping(gamedata, self.raw),
+                'heroComparison': reformat.hero_comparison_mapping(gamedata, self.raw),
+                'player': self.raw['player'],
+            }
+        return self._formatted
+
+    def refresh(self):
+        self._raw = None
+        self._formatted = None
+
+    def to_db_record(self):
+        return {
+            'user': self._id,
+            **self.raw
+        }
+
+    def __gt__(self, other):
+        # since it is gt, other will always be latest_profile,
+        # if it is None, there's no record in db,
+        # Init it
+        if other.raw is None:
+            return True
+        return self.raw['lastUpdate'].__gt__(other.raw['lastUpdate'])
+
+    @property
+    def basic(self):
+        data = self.raw['player']
+        return {
+            'battleTag': data['displayName'],
+            'level': data['level'],
+            'gamewon': data['gameWon'],
+            'endorsement': data['endorsement'],
+            'currentSR': data['ranked']['level'],
+            'highestSR': data['ranked']['highestLevel']
+        }
+
+
+class DBProfile(Profile):
+    def __init__(self, runner):
+        def _retriever(self):
+            return Mongo.db.profile.find_one({
+                'user': self._id
+            }, sort=[
+                ('_id', pymongo.DESCENDING)
+            ])
+        super(DBProfile, self).__init__(runner, _retriever)
+
+
+class RemoteProfile(Profile):
+    def __init__(self, runner):
+        def _retriever(self):
+            return api.get_profile(self.root.credential.bnet_user_cred)
+        super(RemoteProfile, self).__init__(runner, _retriever)

app/bnet_retriever/runner.py

@@ -0,0 +1,30 @@
+from app.bnet_retriever.profile import Profile, RemoteProfile, DBProfile
+from app.bnet_retriever.credential import CredentialManager
+from app.bnet_retriever.stat import CompetitiveStat
+from app.utils.db import Mongo
+from bson import ObjectId
+
+
+class Runner:
+    callbacks = []
+
+    def __init__(self, username: str):
+        record = Mongo.db.user.find_one({'username': username})
+        if not record:
+            raise Exception('No Such User')
+        self.username: str = username
+        self._id: ObjectId = record['_id']
+
+        # Plugs
+        # self.profile: Profile = Profile(self)
+        self.new_profile = RemoteProfile(self)
+        self.latest_profile = DBProfile(self)
+        self.credential: CredentialManager = CredentialManager(self)
+        self.competitive_stat = CompetitiveStat(self)
+
+    def run(self):
+        self.new_profile.refresh()
+        self.latest_profile.refresh()
+        if self.new_profile > self.latest_profile:
+            Mongo.db.profile.insert(self.new_profile.to_db_record())
+            self.competitive_stat.calc()

app/bnet_retriever/stat.py

@@ -0,0 +1,47 @@
+from app.bnet_retriever._plugin import Plugin
+from app.utils.db import Mongo
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class CompetitiveStat(Plugin):
+
+    def get_stats(self):
+        return ({
+            'gamePlayed': round(i['ranked']['所有英雄']['比赛场次']),
+            'won': round(i['ranked']['所有英雄']['比赛胜利']),
+            'lost': round(i['ranked']['所有英雄']['比赛战败']),
+            'draw': round(i['ranked']['所有英雄']['比赛战平']),
+            'skillRate': i['player']['ranked']['level'],
+        } for i in (self.root.latest_profile.formatted['career'], self.root.new_profile.formatted['career']))
+
+    def calc(self):
+        old, new = self.get_stats()
+        if old['gamePlayed'] == new['gamePlayed']:
+            return
+        try:
+            assert new['gamePlayed'] - old['gamePlayed'] == 1, 'GameMissed'
+            sr_diff = new['skillRate'] - old['skillRate']
+            # basically, self.record(sign(sr_diff), sr_diff)
+            # But with assertion
+            if sr_diff > 0:
+                assert new['won'] - old['won'] == 1, 'WonWrong'
+                self.record(1, sr_diff, new)
+            elif sr_diff < 0:
+                assert new['lost'] - old['lost'] == 1, 'LostWrong'
+                self.record(-1, sr_diff, new)
+            elif sr_diff == 0:
+                assert new['draw'] - old['draw'] == 1, 'DrawWrong'
+                self.record(-1, sr_diff, new)
+        except AssertionError as e:
+            logging.error('[%s] calced error, [%s]', self._id, *e.args)
+
+    def record(self, status, sr_diff, stats):
+        Mongo.db.stats.insert({
+            'user': self._id,
+            'time': self.root.latest.raw['lastUpdate'],
+            'gameStatus': status,
+            'skillRateDifference': sr_diff,
+            **stats
+        })

app/bnet_retriever/utils/cookies.py

@@ -0,0 +1,37 @@
+from http.cookiejar import Cookie, CookieJar
+from typing import NewType, Dict, Optional, Tuple, List
+from requests.cookies import RequestsCookieJar
+
+Key = NewType('Key', str)
+Domain = NewType('Domain', str)
+Path = NewType('Path', str)
+
+CookieTuple = Tuple[Key, Domain, Path]
+
+
+class BnetCookieJar(RequestsCookieJar):
+    _template: List[CookieTuple] = [
+        ('_ntes_nuid', '.163.com', '/'),
+        ('MTK_BBID', '.163.com', '/'),
+        ('opt', '.battlenet.com.cn', '/'),
+        ('web.id', '.battlenet.com.cn', '/'),
+        ('BA-tassadar-login.key', '.battlenet.com.cn', '/'),
+        ('login.key', '.battlenet.com.cn', '/'),
+        ('BA-tassadar', '.battlenet.com.cn', '/login'),
+        ('bnet.extra', '.battlenet.com.cn', '/login'),
+    ]
+
+    @classmethod
+    def load_bnet_cookies(cls, obj):
+        new = cls()
+        try:
+            for key, domain, path in cls._template:
+                value = obj[key]
+                new.set(key, value, path=path, domain=domain)
+        except KeyError:
+            raise KeyError('Invalid Cookies: missing keys')
+        return new
+
+    def dump_bnet_cookies(self):
+        tmp = self.get_dict()
+        return {key: tmp[key] for key, *_ in self._template}

app/bnet_retriever/utils/login.py

@@ -0,0 +1,16 @@
+import requests
+from .cookies import BnetCookieJar
+
+
+def login(creds):
+    url = 'https://account.bnet.163.com/battlenet/login?inner_client_id=ow&inner_redirect_uri=http://ow.blizzard.cn/battlenet/login?redirect_url=http://ow.blizzard.cn/career/'
+    sess = requests.Session()
+    jar = BnetCookieJar.load_bnet_cookies(creds)
+    sess.cookies = jar
+    sess.headers.update({
+        'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 13_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148',
+        'Accept-Language': 'en-CN;q=1, zh-Hans-CN;q=0.9, ja-JP;q=0.8',
+        'Accept-Encoding': 'gzip, deflate, br',
+    })
+    sess.get(url)
+    return jar.get('bnet_user_cred'), jar.dump_bnet_cookies()

app/bnet_retriever/utils/reformat.py

@@ -0,0 +1,32 @@
+def achievements_mapping(gamedata, profile):
+    mapping = {item['id']: f"[{category['displayName']}] {item['name']} - {item['description']}" for category in gamedata['achievements'] for item in category['achievements']}
+    return [mapping[i] for i in profile['completedAchievements']]
+
+
+def detailed_mapping(gamedata, profile):
+    heroes_map = {k: v['displayName'] for k, v in gamedata['heroesMap'].items()}
+    # [TODO] Formatting
+    stats_map = {k: v['name'] for k, v in gamedata['stats'].items()}
+    return {
+        i: {
+            heroes_map[k_outer]: {
+                stats_map[k_inner]: v_inner for k_inner, v_inner in v_outer.items()
+            }
+            for k_outer, v_outer in profile['careerStats'][i]['stats'].items()
+        }
+        for i in ['ranked', 'unranked']
+    }
+
+
+def hero_comparison_mapping(gamedata, profile):
+    comparison_name_mapping = {i['id']: i['name'] for i in gamedata['heroComparison']}
+    heroes_map = {k: v['displayName'] for k, v in gamedata['heroesMap'].items()}
+    return {
+        i: {
+            comparison_name_mapping[k]: {
+                heroes_map[item['hero']]: item['value'] for item in v
+            }
+            for k, v in profile['heroComparison'][i].items()
+        }
+        for i in ['ranked', 'unranked']
+    }

app/bnet_retriever/utils/serialize.py

@@ -0,0 +1,11 @@
+import json
+from base64 import b64encode, b64decode
+from typing import Dict
+
+
+def tostr(obj: Dict) -> str:
+    return b64encode(json.dumps(obj).encode()).decode()
+
+
+def fromstr(s: str) -> Dict:
+    return json.loads(b64decode(s.encode()))

app/config.py

@@ -0,0 +1 @@
+MONGO_URL = 'mongodb://127.0.0.1:27017/ow_stats'

app/utils/classproperty.py

@@ -0,0 +1,29 @@
+class ClassPropertyDescriptor(object):
+
+    def __init__(self, fget, fset=None):
+        self.fget = fget
+        self.fset = fset
+
+    def __get__(self, obj, klass=None):
+        if klass is None:
+            klass = type(obj)
+        return self.fget.__get__(obj, klass)()
+
+    def __set__(self, obj, value):
+        if not self.fset:
+            raise AttributeError("can't set attribute")
+        type_ = type(obj)
+        return self.fset.__get__(obj, type_)(value)
+
+    def setter(self, func):
+        if not isinstance(func, (classmethod, staticmethod)):
+            func = classmethod(func)
+        self.fset = func
+        return self
+
+
+def classproperty(func):
+    if not isinstance(func, (classmethod, staticmethod)):
+        func = classmethod(func)
+
+    return ClassPropertyDescriptor(func)

app/utils/db.py

@@ -0,0 +1,21 @@
+from pymongo import MongoClient
+from .classproperty import classproperty
+from app.config import MONGO_URL
+
+
+class Mongo:
+    _db = None
+    _cx = None
+
+    @classproperty
+    def db(cls):
+        if cls._db is None:
+            cls._cx = MongoClient(MONGO_URL)
+            cls._db = cls._cx.get_database()
+        return cls._db
+
+    @classmethod
+    def close(cls):
+        cls._cx.close()
+        cls._db = None
+        cls._cx = None

get_token.py

@@ -0,0 +1,57 @@
+import browser_cookie3 as bc
+import json
+import sys
+import os
+import glob
+from requests.utils import dict_from_cookiejar
+from base64 import b64encode
+from typing import Dict, List, AnyStr
+
+
+def tostr(obj: Dict) -> str:
+    return b64encode(json.dumps(obj).encode()).decode()
+
+
+def generate_path() -> List[AnyStr]:
+    if sys.platform == 'darwin':
+        return glob.glob(os.path.expanduser('~/Library/Application Support/Google/Chrome/Profile 1/Cookies'))
+    elif sys.platform.startswith('linux'):
+        return glob.glob(os.path.expanduser('~/.config/google-chrome/Profile 1/Cookies')) \
+            or glob.glob(os.path.expanduser('~/.config/chromium/Profile 1/Cookies')) \
+            or glob.glob(os.path.expanduser('~/.config/google-chrome-beta/Profile 1/Cookies'))
+    elif sys.platform == 'win32':
+        win_group_policy_path = glob.glob(os.path.join(os.path.split(os.path.split(bc.windows_group_policy_path())[0])[0], 'Profile 1', 'Cookies'))
+        return win_group_policy_path \
+            or glob.glob(os.path.join(os.getenv('APPDATA', ''), '..\Local\\Google\\Chrome\\User Data\\Profile 1\\Cookies')) \
+            or glob.glob(os.path.join(os.getenv('LOCALAPPDATA', ''), 'Google\\Chrome\\User Data\\Profile 1\\Cookies')) \
+            or glob.glob(os.path.join(os.getenv('APPDATA', ''), 'Google\\Chrome\\User Data\\Profile 1\\Cookies'))
+    else:
+        raise NotImplementedError
+
+
+def get_cookies() -> Dict:
+    path = generate_path()
+    _163_cookies = dict_from_cookiejar(bc.chrome(cookie_file=path, domain_name='.163.com'))
+    _bnet_cookies = dict_from_cookiejar(bc.chrome(cookie_file=path, domain_name='.battlenet.com.cn'))
+
+    keys = {
+        '_ntes_nuid',
+        'MTK_BBID',
+        'opt',
+        'web.id',
+        'BA-tassadar-login.key',
+        'login.key',
+        'BA-tassadar',
+        'bnet.extra',
+    }
+
+    return {k: v for k, v in {**_163_cookies, **_bnet_cookies}.items() if k in keys}
+
+
+if __name__ == '__main__':
+    cookies = get_cookies()
+    if not cookies:
+        print('Cookie not found! Please Sign in to Battle.net with Chrome')
+    else:
+        print('Congrats! Here\'s your token: \n')
+        print(tostr(cookies))

requirements.py

@@ -0,0 +1,6 @@
+certifi==2019.6.16
+chardet==3.0.4
+idna==2.8
+pymongo==3.8.0
+requests==2.22.0
+urllib3==1.25.3