2 changed files with 125 additions and 0 deletions
@ -0,0 +1,91 @@
@@ -0,0 +1,91 @@
|
||||
#! /usr/bin/python3 |
||||
# -*- coding:utf-8 -*- |
||||
|
||||
from src import random_selector |
||||
from src import main |
||||
from src import lottery |
||||
import functools |
||||
import importlib |
||||
|
||||
_GLOBAL_FUNC_MAPS = {} |
||||
_GLOBAL_EXCLUDE_ROLE_LIST = set() |
||||
|
||||
|
||||
def exclude_users(users, exclude_roles): |
||||
users[:] = [u for u in users if u['role'] not in exclude_roles] |
||||
return users |
||||
|
||||
|
||||
def before_unbound_method(attr_name): |
||||
def decorator(func): |
||||
@functools.wraps(func) |
||||
def wrapper(*args, **kwargs): |
||||
result = func(*args, **kwargs) |
||||
old_attr = _GLOBAL_FUNC_MAPS.get(attr_name) |
||||
if callable(old_attr): |
||||
return old_attr(*args, **kwargs) |
||||
else: |
||||
return result |
||||
return wrapper |
||||
return decorator |
||||
|
||||
|
||||
def after_unbound_method(attr_name): |
||||
def decorator(func): |
||||
@functools.wraps(func) |
||||
def wrapper(*args, **kwargs): |
||||
old_attr = _GLOBAL_FUNC_MAPS.get(attr_name) |
||||
if callable(old_attr): |
||||
result = old_attr(*args, **kwargs) |
||||
func(*args, **kwargs) |
||||
else: |
||||
result = func(*args, **kwargs) |
||||
return result |
||||
return wrapper |
||||
return decorator |
||||
|
||||
|
||||
def replace_unbound_method(ins, attr_name, value): |
||||
old_attr = getattr(ins, attr_name) |
||||
setattr(ins, attr_name, value) |
||||
return old_attr |
||||
|
||||
|
||||
@before_unbound_method('randomselect') |
||||
def on_randomselect(elapsed_ms, items, select_num): |
||||
exclude_users(items, _GLOBAL_EXCLUDE_ROLE_LIST) |
||||
|
||||
|
||||
@main.app.route('/exclude/<role:int>') |
||||
def exclude_role(role): |
||||
_GLOBAL_EXCLUDE_ROLE_LIST.add(role) |
||||
return lottery.server_resp(200, "success") |
||||
|
||||
|
||||
@main.app.route('/include/<role:int>') |
||||
def include_role(role): |
||||
try: |
||||
_GLOBAL_EXCLUDE_ROLE_LIST.remove(role) |
||||
except KeyError: |
||||
return lottery.server_resp(501, "找不到该角色id", {'role': role}) |
||||
else: |
||||
return lottery.server_resp(200, "success") |
||||
|
||||
|
||||
@main.app.route('/show_exclude') |
||||
def show_exclude(): |
||||
return lottery.server_resp(200, "success", [r for r in _GLOBAL_EXCLUDE_ROLE_LIST]) |
||||
|
||||
|
||||
def patch(): |
||||
_GLOBAL_FUNC_MAPS['randomselect'] = replace_unbound_method(importlib.reload(random_selector), |
||||
'randomselect', on_randomselect) |
||||
|
||||
|
||||
def run(host, port, debug): |
||||
patch() |
||||
main.run(host=host, port=port, debug=debug) |
||||
|
||||
|
||||
if __name__ == '__main__': |
||||
run('0.0.0.0', 8080, False) |
||||
@ -0,0 +1,34 @@
@@ -0,0 +1,34 @@
|
||||
#!/usr/bin/python3 |
||||
# -*- coding:utf-8 -*- |
||||
|
||||
import json |
||||
|
||||
|
||||
def write_db(db, filepath): |
||||
|
||||
try: |
||||
json.dump(db, open(filepath, 'w', encoding='utf8')) |
||||
except (json.JSONDecodeError, IOError) as e: |
||||
print(e.args) |
||||
raise |
||||
|
||||
|
||||
def text_to_db(txt_path): |
||||
with open(txt_path, 'r', encoding='utf8') as txt_fb: |
||||
db = [] |
||||
while True: |
||||
line = txt_fb.readline() |
||||
if not line: |
||||
break |
||||
line = line.strip() |
||||
db.append({'uid': line, 'role': 1, 'name': line}) |
||||
return db |
||||
|
||||
|
||||
def main(txt_path, db_path): |
||||
write_db(text_to_db(txt_path), db_path) |
||||
print('生成成功') |
||||
|
||||
|
||||
if __name__ == "__main__": |
||||
main('users.txt', r'doc\users.json') |
||||
Loading…
Reference in new issue