Browse Source

initial commit

master
BotoX 1 year ago
commit
30bb08712a
  1. 1
      .flaskenv
  2. 6
      .gitignore
  3. 17
      WSGI.py
  4. 11
      app.py
  5. 4
      config.py
  6. 9
      db_create.py
  7. 0
      demweb/__init__.py
  8. 85
      demweb/app.py
  9. 10
      demweb/extensions.py
  10. 84
      demweb/models.py
  11. 0
      done.txt
  12. 1
      oofsgi/.gitignore
  13. 29
      oofsgi/uwsgi.ini
  14. 434
      parser.py
  15. 16
      requirements.txt
  16. 33939
      work.txt

1
.flaskenv

@ -0,0 +1 @@ @@ -0,0 +1 @@
FLASK_APP=app.py

6
.gitignore vendored

@ -0,0 +1,6 @@ @@ -0,0 +1,6 @@
# python
__pycache__/
venv/
# editor
.vscode/

17
WSGI.py

@ -0,0 +1,17 @@ @@ -0,0 +1,17 @@
#!/usr/bin/python3
# -*- coding: utf-8 -*-
import gevent.monkey
gevent.monkey.patch_all()
from demweb.app import create_app
app = create_app()
if app.config['DEBUG']:
from werkzeug.debug import DebuggedApplication
app.wsgi_app = DebuggedApplication(app.wsgi_app, True)
if __name__ == '__main__':
import gevent.pywsgi
gevent_server = gevent.pywsgi.WSGIServer(("localhost", 5000), app.wsgi_app)
gevent_server.serve_forever()

11
app.py

@ -0,0 +1,11 @@ @@ -0,0 +1,11 @@
#!/usr/bin/env python3
import gevent.monkey # noqa isort:skip
gevent.monkey.patch_all() # noqa isort:skip
from demweb.app import create_app
if __name__ == '__main__':
app = create_app()
host = '0.0.0.0'
port = 5000
app.run(host, port)

4
config.py

@ -0,0 +1,4 @@ @@ -0,0 +1,4 @@
# pylint: disable=line-too-long
SQLALCHEMY_DATABASE_URI = 'mysql://demweb:demweb@127.0.0.1:3306/demweb?charset=utf8mb4'
SQLALCHEMY_TRACK_MODIFICATIONS = False

9
db_create.py

@ -0,0 +1,9 @@ @@ -0,0 +1,9 @@
#!/usr/bin/env python3
from demweb.app import create_app
from demweb.extensions import db
if __name__ == '__main__':
app = create_app()
with app.app_context():
db.create_all()
db.session.commit()

0
demweb/__init__.py

85
demweb/app.py

@ -0,0 +1,85 @@ @@ -0,0 +1,85 @@
from contextlib import nullcontext
from datetime import datetime
from flask import Flask, g, has_app_context, request
from .extensions import default_config
def register_extensions(app):
from .extensions import db
db.init_app(app)
def register_blueprints(app):
with app.app_context():
# Register blueprints
pass
def setup_logging(app):
# Logging
if app.debug:
@app.before_request
def log_before_request():
g.log_datetime = datetime.now()
@app.after_request
def log_after_request(response):
now = datetime.now()
delta = (now - g.log_datetime).total_seconds() * 1_000
dt = now.isoformat().replace('T', ' ').split('.')[0]
print(f'[{dt}] {request.method} {request.path} -> {response.status} ({delta:.2f}ms)')
return response
def register_shell_context(app):
import inspect
from pprint import pprint
from flask_sqlalchemy import get_debug_queries
from demweb import models
from demweb.extensions import db
@app.shell_context_processor
def make_shell_context():
return {
'app': app,
'db': db,
'query': get_debug_queries,
'print': pprint,
**dict(inspect.getmembers(models, inspect.isclass))
}
def fix_sqlalchemy_uwsgi_multiprocess_bug(app):
from demweb.extensions import db
def _dispose_db_pool():
print('uWSGI+SQLAlchemy: Disposing forked() db pool!')
with app.app_context() if not has_app_context() else nullcontext():
db.engine.dispose()
try:
from uwsgidecorators import postfork
postfork(_dispose_db_pool)
except ImportError:
pass
def create_app(test_config=None):
# Create flask application object
app = Flask(__name__)
if test_config is None:
app.config.update(default_config)
else:
app.config.update(test_config)
register_extensions(app)
register_blueprints(app)
register_shell_context(app)
setup_logging(app)
fix_sqlalchemy_uwsgi_multiprocess_bug(app)
return app

10
demweb/extensions.py

@ -0,0 +1,10 @@ @@ -0,0 +1,10 @@
from flask_sqlalchemy import SQLAlchemy
import config
db = SQLAlchemy()
default_config = dict()
for key in dir(config):
if key.isupper():
default_config[key] = getattr(config, key)

84
demweb/models.py

@ -0,0 +1,84 @@ @@ -0,0 +1,84 @@
# pylint: disable=no-member
from flask import current_app
from sqlalchemy.types import JSON
from demweb.extensions import db
class Session(db.Model):
id = db.Column(db.Integer, primary_key=True)
time = db.Column(db.DateTime, index=True)
length = db.Column(db.Float, index=True)
demoname = db.Column(db.String(255), unique=True)
mapname = db.Column(db.String(255))
mapmd5 = db.Column(db.String(32))
servername = db.Column(db.String(255))
frames = db.Column(db.Integer)
ticks = db.Column(db.Integer)
tickinterval = db.Column(db.Float)
dirty = db.Column(db.Boolean)
playtime = db.Column(db.Float, index=True)
rounds = db.Column(db.Integer, index=True)
ct_wins = db.Column(db.Integer, index=True)
t_wins = db.Column(db.Integer, index=True)
chats = db.Column(db.Integer, index=True)
deaths = db.Column(db.Integer, index=True)
kills = db.Column(db.Integer, index=True)
voice_active = db.Column(db.Float, index=True)
voice_total = db.Column(db.Float, index=True)
silence_chunks = db.Column(db.JSON(65535))
class Player(db.Model):
guid = db.Column(db.String(32), primary_key=True)
name = db.Column(db.String(32), primary_key=True)
first_seen = db.Column(db.DateTime, index=True)
last_seen = db.Column(db.DateTime, index=True)
playtime = db.Column(db.Float, index=True)
chats = db.Column(db.Integer, index=True)
deaths = db.Column(db.Integer, index=True)
kills = db.Column(db.Integer, index=True)
voicetime = db.Column(db.Float, index=True)
class PlayerNames(db.Model):
guid = db.Column(db.String(32), primary_key=True)
name = db.Column(db.String(32), primary_key=True)
time = db.Column(db.Float)
class PlayerSprays(db.Model):
guid = db.Column(db.String(32), primary_key=True)
spray = db.Column(db.String(32), primary_key=True)
class PlayerSession(db.Model):
player_guid = db.Column(db.String(32), db.ForeignKey('player.guid'), primary_key=True)
session_id = db.Column(db.Integer, db.ForeignKey('session.id'), primary_key=True)
playtime = db.Column(db.Float)
chats = db.Column(db.Integer)
deaths = db.Column(db.Integer)
kills = db.Column(db.Integer)
voicetime = db.Column(db.Float)
voice_chunks = db.Column(db.JSON(65535))
class Chat(db.Model):
id = db.Column(db.Integer, primary_key=True)
player_guid = db.Column(db.String(32), db.ForeignKey('player.guid'))
session_id = db.Column(db.Integer, db.ForeignKey('session.id'))
time = db.Column(db.DateTime, index=True)
name = db.Column(db.String(255))
chat = db.Column(db.String(1024))
class Event(db.Model):
id = db.Column(db.Integer, primary_key=True)
player_guid = db.Column(db.String(32), db.ForeignKey('player.guid'))
session_id = db.Column(db.Integer, db.ForeignKey('session.id'))
time = db.Column(db.DateTime, index=True)
event = db.Column(db.String(32), index=True)
data = db.Column(db.JSON(1024))

1
oofsgi/.gitignore vendored

@ -0,0 +1 @@ @@ -0,0 +1 @@
uwsgi.sock

29
oofsgi/uwsgi.ini

@ -0,0 +1,29 @@ @@ -0,0 +1,29 @@
[uwsgi]
# emperor
chdir = ..
# socket = [addr:port]
socket = oofsgi/uwsgi.sock
chmod-socket = 664
# WSGI module and callable
# module = [wsgi_module_name]:[application_callable_name]
module = WSGI:app
# master = [master process (true of false)]
master = true
# debugging
catch-exceptions = True
# disable request logging
disable-logging=True
# performance
processes = 4
buffer-size = 8192
# async
loop = gevent
gevent = 2048
gevent-monkey-patch = true

434
parser.py

@ -0,0 +1,434 @@ @@ -0,0 +1,434 @@
#!/usr/bin/env python3
import sys
import orjson
import time
import os.path
import multiprocessing
import signal
import re
import traceback
import subprocess
from datetime import datetime, timedelta
from sqlalchemy import exc, func
from sqlalchemy.sql.expression import bindparam
from flask import current_app
from demweb.app import create_app
from demweb.extensions import db
from demweb import models
DEMO_PATH_PREFIX = ''
DEMO_REGEX = re.compile(r'auto-(\d+-\d+)-(\w+)')
def _itery(x):
if x is None:
return ()
if isinstance(x, list):
return x
return (x,)
def remove_chatcolors(message):
output = ''
msglen = len(message)
i = 0
while i < msglen:
c = message[i]
if ord(c) <= 8:
if ord(c) == 7:
i += 6
elif ord(c) == 8:
i += 8
else:
output += c
i += 1
return output
silence_start_re = re.compile(' silence_start: (?P<start>[0-9]+(\.?[0-9]*))$')
silence_end_re = re.compile(' silence_end: (?P<end>[0-9]+(\.?[0-9]*)) ')
total_duration_re = re.compile('size=[^ ]+ time=(?P<hours>[0-9]{2}):(?P<minutes>[0-9]{2}):(?P<seconds>[0-9\.]{5}) bitrate=')
def get_chunk_times(path):
proc = subprocess.Popen([
'ffmpeg',
'-i', path,
'-af', 'silencedetect=n=-80dB:d=0.1',
'-f', 'null',
'-'],
stderr=subprocess.PIPE,
stdout=subprocess.DEVNULL
)
output = proc.communicate()[1].decode('utf-8')
lines = output.splitlines()
# Chunks start when silence ends, and chunks end when silence starts.
chunk_starts = []
chunk_ends = []
for line in lines:
silence_start_match = silence_start_re.search(line)
silence_end_match = silence_end_re.search(line)
total_duration_match = total_duration_re.search(line)
if silence_start_match:
start = float(silence_start_match.group('start'))
if start == 0.:
# Ignore initial silence.
continue
chunk_ends.append(start)
if len(chunk_starts) == 0:
# Started with non-silence.
chunk_starts.append(0.)
elif silence_end_match:
chunk_starts.append(float(silence_end_match.group('end')))
elif total_duration_match:
hours = int(total_duration_match.group('hours'))
minutes = int(total_duration_match.group('minutes'))
seconds = float(total_duration_match.group('seconds'))
end_time = hours * 3600 + minutes * 60 + seconds
if len(chunk_starts) == 0:
# No silence found.
chunk_starts.append(0)
if len(chunk_starts) > len(chunk_ends):
if abs(chunk_starts[-1] - end_time) < 0.1:
# Last chunk starts at very the end? nah.
del chunk_starts[-1]
else:
# Finished with non-silence.
chunk_ends.append(end_time)
return list(zip(chunk_starts, chunk_ends))
def parse_demo(path):
jmodified = False
with open(DEMO_PATH_PREFIX + path + '/out.json', 'r') as fp:
data = orjson.loads(fp.read())
demoname = os.path.basename(path)
match = DEMO_REGEX.match(demoname)
starttime = datetime.strptime(match.group(1), '%Y%m%d-%H%M%S')
tickinterval = data['serverinfo']['tickInterval']
session = models.Session(
time=starttime,
length=data['demoheader']['playback_time'],
demoname=demoname,
mapname=match.group(2),
mapmd5=data['serverinfo']['mapMD5'],
servername=data['demoheader']['servername'],
frames=data['demoheader']['playback_frames'],
ticks=data['demoheader']['playback_ticks'],
tickinterval=tickinterval,
dirty=data['demoheader']['dirty'],
playtime=0,
rounds=0,
ct_wins=0,
t_wins=0,
chats=0,
deaths=0,
kills=0,
voice_active=data['voice']['active_time'],
voice_total=data['voice']['total_time'],
silence_chunks=data['voice']['silence'] if 'silence' in data['voice'] else []
)
db.session.add(session)
db.session.commit()
session_playtime = 0
session_chats = 0
session_deaths = 0
session_kills = 0
if not data['players']:
data['players'] = dict()
for guid, obj in data['players'].items():
# convert to seconds
obj['playtime'] *= tickinterval
# try insert new player
while True:
try:
db.session.execute(
models.Player.__table__.insert().prefix_with('IGNORE').values(
guid=guid,
name=None,
first_seen=starttime,
last_seen=starttime,
playtime=0,
chats=0,
deaths=0,
kills=0,
voicetime=0
))
db.session.commit()
break
except Exception:
db.session.rollback()
print('deadlock 1')
# update player stats atomically
while True:
try:
db.session.execute(
models.Player.__table__.update(models.Player.guid == guid).values(
playtime = models.Player.playtime + obj['playtime'],
chats = models.Player.chats + obj['chats'],
deaths = models.Player.deaths + obj['deaths'],
kills = models.Player.kills + obj['kills'],
voicetime = models.Player.voicetime + obj['voicetime'],
first_seen = func.least(models.Player.first_seen, starttime),
last_seen = func.greatest(models.Player.first_seen, starttime),
))
db.session.commit()
break
except Exception:
db.session.rollback()
print('deadlock 2')
# try insert new player names
while True:
try:
db.session.execute(
models.PlayerNames.__table__.insert().prefix_with('IGNORE').values([
dict(
guid = guid,
name = name,
time = 0
)
for name in obj['names'].keys()
]))
db.session.commit()
break
except Exception:
db.session.rollback()
print('deadlock 3')
# update player names atomically
while True:
try:
db.session.execute(
models.PlayerNames.__table__.update()
.where(models.PlayerNames.guid == bindparam('_guid'))
.where(models.PlayerNames.name == bindparam('_name'))
.values(time = models.PlayerNames.time + bindparam('_nametime')),
[
dict(
_guid = guid,
_name = name,
_nametime = nametime
) for name, nametime in obj['names'].items()
]
)
db.session.commit()
break
except Exception:
db.session.rollback()
print('deadlock 4')
# try insert new player sprays
while True:
try:
if obj['sprays']:
db.session.execute(
models.PlayerSprays.__table__.insert().prefix_with('IGNORE').values(
[(guid, spray) for spray in obj['sprays']]
))
db.session.commit()
break
except Exception:
db.session.rollback()
print('deadlock 5')
# calculate player voice chunks/timing if possible and not exists
if obj['voicetime'] > 0 and 'voice_chunks' not in obj:
voicefile = DEMO_PATH_PREFIX + path + '/voice/' + guid + '.opus'
if os.path.isfile(voicefile):
obj['voice_chunks'] = get_chunk_times(voicefile)
jmodified = True
# insert new player session
player_session = models.PlayerSession(
player_guid=guid,
session_id=session.id,
playtime=obj['playtime'],
chats=obj['chats'],
deaths=obj['deaths'],
kills=obj['kills'],
voicetime=obj['voicetime'],
voice_chunks=obj['voice_chunks'] if 'voice_chunks' in obj else None
)
db.session.add(player_session)
db.session.commit()
if guid != "BOT":
session_playtime += player_session.playtime
session_chats += player_session.chats
session_deaths += player_session.deaths
session_kills += player_session.kills
session.playtime = session_playtime
session.chats = session_chats
session.deaths = session_deaths
session.kills = session_kills
db.session.commit()
for obj in _itery(data['events']):
if obj['event'] == 'round_start':
session.rounds += 1
elif obj['event'] == 'round_end':
if obj['winner'] == 2:
session.t_wins += 1
elif obj['winner'] == 3:
session.ct_wins += 1
event = models.Event(
player_guid=guid,
session_id=session.id,
time=starttime + timedelta(seconds=obj['tick'] / tickinterval),
event=obj['event'],
data=obj
)
db.session.add(event)
db.session.commit()
for obj in _itery(data['chat']):
msg = obj['msgName']
if msg == '#Cstrike_Name_Change':
nick = obj['msgSender']
msg = 'changed their name to "{}"'.format(obj['msgText'])
elif obj['steamid'] == 'BOT':
nick = 'Console'
msg = obj['msgName'].lstrip('\x01\x07FF0000Console: ')
else:
try:
nick, msg = obj['msgName'][1:].split('\x01', 1)
except Exception:
msg = None
if msg:
msg = msg.lstrip(': ')
if msg:
nick = remove_chatcolors(nick)
msg = remove_chatcolors(msg)
chat = models.Chat(
player_guid=obj['steamid'],
session_id=session.id,
time=starttime + timedelta(seconds=obj['tick'] / tickinterval),
name=nick,
chat=msg
)
db.session.add(chat)
db.session.commit()
if jmodified:
with open(DEMO_PATH_PREFIX + path + '/out.json', 'wb') as fp:
fp.write(orjson.dumps(data, option=orjson.OPT_INDENT_2))
def worker(queue, error_queue):
app = create_app()
with app.app_context():
while True:
item = queue.get()
if not item:
queue.task_done()
break
try:
db.session.commit()
parse_demo(item)
except Exception as e:
error_queue.put(item)
print(f'vvv error from: {item} vvv')
traceback.print_exception(type(e), e, e.__traceback__)
print(f'^^^ error from: {item} ^^^')
queue.task_done()
if __name__ == '__main__':
if len(sys.argv) >= 2:
DEMO_PATH_PREFIX = sys.argv[1]
with open('work.txt', 'r') as fp:
work = set(fp.read().splitlines())
with open('done.txt', 'r') as fp:
done = set(fp.read().splitlines())
if True:
done = set()
app = create_app()
with app.app_context():
meta = db.metadata
for table in reversed(meta.sorted_tables):
db.session.execute(table.delete())
db.session.commit()
todo = work - done
q = multiprocessing.JoinableQueue()
q_err = multiprocessing.SimpleQueue()
# populate queue with jobs
print('Populating queue...')
for task in todo:
q.put(task)
# ignore STDINT(^C) and store original sigint handler
# this is so our forked processes ignore ^C and can finish their work cleanly
original_sigint_handler = signal.signal(signal.SIGINT, signal.SIG_IGN)
# start workers
print('Starting workers...')
workers = []
for i in range(20):
p = multiprocessing.Process(target=worker, args=(q, q_err))
workers.append(p)
p.start()
# restore original sigint handler (don't ignore ^C)
signal.signal(signal.SIGINT, original_sigint_handler)
# wait for queue to empty or ^C
print('Waiting for jobs to complete or ^C...')
try:
q.join()
except KeyboardInterrupt:
print('Quit!')
# get any jobs which were left in the queue
print('Emptying queue...')
left = set()
while not q.empty():
item = q.get()
left.add(item)
q.task_done()
# send N kill signals for each worker
print('Putting kill jobs into queue...')
for _ in workers:
q.put(None)
q.close()
# wait until workers are dead
print('Waiting for workers to finish...')
for w in workers:
w.join()
# get the failed jobs
print('Emptying failed queue...')
while not q_err.empty():
item = q_err.get()
left.add(item)
q_err.close()
done.update(todo - left)
with open('done.txt', 'w') as fp:
fp.write('\n'.join(done))

16
requirements.txt

@ -0,0 +1,16 @@ @@ -0,0 +1,16 @@
click==8.0.0
Flask==2.0.0
Flask-SQLAlchemy==2.5.1
gevent==21.1.2
greenlet==1.1.0
itsdangerous==2.0.0
Jinja2==3.0.0
MarkupSafe==2.0.0
mysqlclient==2.0.3
orjson==3.5.2
python-dotenv==0.17.1
SQLAlchemy==1.4.15
tqdm==4.60.0
Werkzeug==2.0.0
zope.event==4.5.0
zope.interface==5.4.0

33939
work.txt

File diff suppressed because it is too large Load Diff
Loading…
Cancel
Save