ops/ops_interface
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

gamlog config 更新接口

Browse Source
This commit is contained in:
pengtao 2019-10-12 15:27:13 +08:00
parent c5533a97c1
commit 6d9997a72f
9 changed files with 602 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
.idea/vcs.xml generated Normal file
View File

@ -0,0 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="$PROJECT_DIR$" vcs="Git" />
</component>
</project>

10
README.md
View File

@ -0,0 +1,10 @@
#### mp 合作方特征码接口
从NP系统获取外部特征码，匹配中文用于数数展示
externalkey2gamelog
#### gamelog appkey 写入接口
从接口获取appkey写入gamelog配置文件

14
config/config.py Normal file
View File

@ -0,0 +1,14 @@
# -*- coding: utf-8 -*-
import os
ad_env = os.getenv('ad_env') or 'prod'
log_path = "/data/logs"
if ad_env == 'prod':
interface_port = 6500
elif ad_env == 'test':
interface_port = 6600
else:
raise Exception("GET config with ad_env failed!")
if not interface_port:
interface_port = 6500

94
data_channel/gamelog_external.py Normal file
View File

@ -0,0 +1,94 @@
# -*- coding: utf-8 -*-
import requests
import logging
from ops.mansible import AnsibleAPI
from ops.mmysql import MysqlBase
import time
log = logging.getLogger(__name__)
class Build_Gamelog_Config():
def __init__(self):
self.url = "https://mp.kingsome.cn/api/open/games/keys"
self.filename = "/tmp/gamelog.appkey.php"
self.dest_file = "/data/apps/gamelog_php/config/gamelog.appkey.php"
self.dest_paths = "/data/apps/gamelog_php/config/"
self.project = "gamelog_php"
def get_keys(self):
r = requests.get(self.url)
if r.status_code == requests.codes.ok:
return r.json()
else:
return None
def split_data(self, data):
out = []
if data:
try:
for line in data['gameList']:
out.append(line)
except Exception:
log.error(f"split {data} failed", exc_info=True)
return out
def build_config(self, data):
start = "<?php"
end = ");"
content = ""
if data:
for line in data:
try:
content += f"""return array(
"{line['game_id']}" => array(
'gamename' => '{line['game_name']}',
'appkey' => '{line['app_key']}',
)"""
except Exception:
log.error(f"build context failed with {line}.", exc_info=True)
with open(self.filename, 'w') as f:
f.write(f"{start}+'\n'+{content}+'\n'+{end}")
def deploy_config(self):
ip_addr = self.get_ip_addr()
hostfile = self.write_host(ip_addr)
an = AnsibleAPI(hostfile)
data = {'dest_paths': self.dest_paths, 'source': self.filename, 'dest_filename': self.dest_file}
resule = an.run_playbook('/data/git/dev_ops/ansible/playbook/publish_file2server.yml', **data)
if not (resule['failed'] or resule['unreachable']):
log.info(f"deploy success!")
return True
else:
log.error(f"deploy failed info={resule}")
return False
def get_ip_addr(self):
ip_addr = list()
sql = f"SELECT run_server FROM `deploy_projectmanage` WHERE project_name='{self.project}' AND env='prod'"
args = dict()
args['host'] = '10.10.3.5'
args['user'] = 'deploy'
args['pswd'] = 'deploy2016'
args['db'] = 'ywplatform'
mdb = MysqlBase(**args)
data = mdb.query(sql)
if data:
for line in data[0]:
ip_addr.append(line)
return ip_addr
def write_host(self, ip_addr):
host_file = "%s/host%s" % ('/tmp', time.strftime("%Y%m%d%H%M%s"))
with open(host_file, 'a+') as f:
for x in ip_addr:
f.writelines(x + '\n')
return host_file
def run(self):
keys = self.get_keys()
data = self.split_data(keys)
self.build_config(data)
self.deploy_config()

5
data_channel/mp2shushu.py Normal file
View File

@ -0,0 +1,5 @@
# -*- coding: utf-8 -*-
class Mp2shushu():
def __init__(self, channel, gameid):
self.mp_url = f"https://mp.kingsome.cn/api/open/promotion/co-list?channelid={channel}&gameid={gameid}"

221
ops/mansible.py Normal file
View File

@ -0,0 +1,221 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import tempfile
from collections import namedtuple
from ansible.parsing.dataloader import DataLoader
from ansible.vars.manager import VariableManager
from ansible.inventory.manager import InventoryManager
from ansible.playbook.play import Play
from ansible.executor.playbook_executor import PlaybookExecutor
from ansible.executor.task_queue_manager import TaskQueueManager
from ansible.plugins.callback import CallbackBase
import json
import pdb
import copy
class ResultsCallback(CallbackBase):
def __init__(self, *args, **kwargs):
super(ResultsCallback, self).__init__(*args, **kwargs)
self.task_ok = {}
self.task_unreachable = {}
self.task_failed = {}
self.task_skipped = {}
self.task_stats = {}
# self.host_ok = {}
# self.host_unreachable = {}
# self.host_failed = {}
def v2_runner_on_unreachable(self, result):
self.task_unreachable[result._host.get_name()] = result
def v2_runner_on_ok(self, result, *args, **kwargs):
self.task_ok[result._host.get_name()] = result
def v2_runner_on_failed(self, result, *args, **kwargs):
self.task_failed[result._host.get_name()] = result
def v2_runner_on_skipped(self, result, *args, **kwargs):
self.task_skipped[result._host.get_name()] = result
def v2_runner_on_stats(self, result, *args, **kwargs):
self.task_stats[result._host.get_name()] = result
class AnsibleAPI(object):
def __init__(self, hostfile, *args, **kwargs):
self.loader = DataLoader()
self.results_callback = ResultsCallback()
# if not os.path.isfile(hostfile):
# raise Exception("%s file not found!" % hostfile)
self.inventory = InventoryManager(loader=self.loader, sources=[hostfile])
self.variable_manager = VariableManager(loader=self.loader, inventory=self.inventory)
self.passwords = None
Options = namedtuple('Options',
['connection',
'remote_user',
'ask_sudo_pass',
'verbosity',
'ack_pass',
'module_path',
'forks',
'become',
'become_method',
'become_user',
'check',
'listhosts',
'listtasks',
'listtags',
'syntax',
'sudo_user',
'sudo',
'diff'])
# 初始化需要的对象
self.options = Options(connection='smart',
remote_user=None,
ack_pass=None,
sudo_user=None,
forks=5,
sudo=None,
ask_sudo_pass=False,
verbosity=5,
module_path=None,
become=None,
become_method=None,
become_user=None,
check=False,
diff=False,
listhosts=None,
listtasks=None,
listtags=None,
syntax=None)
@staticmethod
def deal_result(info):
host_ips = list(info.get('success').keys())
info['success'] = host_ips
error_ips = info.get('failed')
error_msg = {}
for key, value in error_ips.items():
temp = {}
temp[key] = value.get('stderr')
error_msg.update(temp)
info['failed'] = error_msg
return info
# return json.dumps(info)
def run(self, module_name, module_args):
play_source = dict(
name="Ansible Play",
hosts='all',
gather_facts='no',
tasks=[
dict(action=dict(module=module_name, args=module_args), register='shell_out'),
# dict(action=dict(module='debug', args=dict(msg='{{shell_out.stdout}}')))
]
)
play = Play().load(play_source, variable_manager=self.variable_manager, loader=self.loader)
tqm = None
try:
tqm = TaskQueueManager(
inventory=self.inventory,
variable_manager=self.variable_manager,
loader=self.loader,
options=self.options,
passwords=self.passwords,
stdout_callback=self.results_callback,
)
result = tqm.run(play)
finally:
if tqm is not None:
tqm.cleanup()
##定义字典用于接收或者处理结果
result_raw = {'success': {}, 'failed': {}, 'unreachable': {}, 'skipped': {}, 'status': {}}
# 循环打印这个结果，success，failed，unreachable需要每个都定义一个
for host, result in self.results_callback.task_ok.items():
result_raw['success'][host] = result._result
for host, result in self.results_callback.task_failed.items():
result_raw['failed'][host] = result._result
for host, result in self.results_callback.task_unreachable.items():
result_raw['unreachable'][host] = result._result
result_full = copy.deepcopy(result_raw)
result_json = self.deal_result(result_raw)
if not result_json['failed']:
return result_json
else:
return result_full
# return result_raw
def run_playbook(self, file, **kwargs):
if not os.path.isfile(file):
raise Exception("%s file not found!" % file)
try:
# extra_vars = {} # 额外的参数 sudoers.yml以及模板中的参数，它对应ansible-playbook test.yml --extra-vars "host='aa' name='cc' "
self.variable_manager.extra_vars = kwargs
playbook = PlaybookExecutor(playbooks=['' + file], inventory=self.inventory,
variable_manager=self.variable_manager,
loader=self.loader, options=self.options, passwords=self.passwords)
playbook._tqm._stdout_callback = self.results_callback
playbook.run()
except Exception as e:
print("error:", e.message)
##定义字典用于接收或者处理结果
result_raw = {'success': {}, 'failed': {}, 'unreachable': {}, 'skipped': {}, 'status': {}}
# 循环打印这个结果，success，failed，unreachable需要每个都定义一个
for host, result in self.results_callback.task_ok.items():
result_raw['success'][host] = result._result
for host, result in self.results_callback.task_failed.items():
result_raw['failed'][host] = result._result
for host, result in self.results_callback.task_unreachable.items():
result_raw['unreachable'][host] = result._result
# for host, result in self.results_callback.task_skipped.items():
# result_raw['skipped'][host] = result._result
#
# for host, result in self.results_callback.task_stats.items():
# result_raw['status'][host] = result._result
result_full = copy.deepcopy(result_raw)
result_json = self.deal_result(result_raw)
if not result_json['failed']:
return result_json
else:
return result_full
class AnsiInterface(AnsibleAPI):
def __init__(self, hostfile, *args, **kwargs):
super(AnsiInterface, self).__init__(hostfile, *args, **kwargs)
def copy_file(self, src=None, dest=None):
"""
copy file
"""
module_args = "src=%s dest=%s" % (src, dest)
result = self.run('copy', module_args)
return result
def exec_command(self, cmds):
"""
commands
"""
result = self.run('command', cmds)
return result
def exec_script(self, path):
"""
在远程主机执行shell命令或者.sh脚本
"""
result = self.run('shell', path)
return result

173
ops/mmysql.py Normal file
View File

@ -0,0 +1,173 @@
#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
date: 2016/07/11
role: mysql的增删改查类
usage: m = mysqlBase(host='xxx',db='xxx',user='xxx',pwd='xxx') 实例化
m.insert('core',{'host_name':'ccc','process_name':'ddd','ip_addr':'192.168.136.41','status':'4'})
m.update('table',{'field1':'value1','field2':'value2'},'id=1') 更新表名 字段名：值 条件
m.delete('core','status=5 and id=12')
m.change("update core set a='aaa' where id=1") 可以多条插入
m.query("select * from core")
"""
from .mlog import log
import warnings
try:
import MySQLdb
except:
import pymysql
pymysql.install_as_MySQLdb()
import MySQLdb
###mysql操作类
class MysqlBase:
###连接数据库
def __init__(self, **args):
###获取参数
self.host = args.get('host', 'localhost')
self.user = args.get('user')
self.pswd = args.get('pswd')
self.db = args.get('db', 'mysql')
self.port = args.get('port', '3306')
self.charset = args.get('charset', 'utf8')
try:
self.conn = MySQLdb.connect(host=self.host, user=self.user, passwd=self.pswd, db=self.db,
port=int(self.port), charset=self.charset)
self.curs = self.conn.cursor()
self.curs.execute('SET NAMES utf8')
except:
log.error('%s mysql connect error' % self.host)
raise ValueError('mysql connect error %s' % self.host)
###释放资源
def __del__(self):
self.curs.close()
self.conn.close()
###插入
def insert(self, table, data):
_field = ','.join(['`%s`' % (k_insert) for k_insert in data.keys()])
_value = ','.join(["'%s'" % (str(v_insert).replace("'", "\'")) for v_insert in data.values()])
###拼接成sql语句
_sql = 'INSERT INTO `%s`(%s) VALUES(%s)' % (table, _field, _value)
###执行
self.curs.lastrowid = 0
try:
self.curs.execute(_sql)
###提交
self.conn.commit()
#log.info('%s insert ' % _sql)
except:
self.conn.rollback()
log.error('%s insert error' % _sql)
raise ValueError('112,insert error %s' % _sql)
return self.curs.lastrowid
###更新
def update(self, table, data, condition):
_field = ','.join(["`%s`='%s'" % (k_update, str(data[k_update]).replace("'", "\'")) for k_update in data])
_sql = 'UPDATE `%s` SET %s WHERE %s' % (table, _field, condition)
###执行
resNum = 0
try:
resNum = self.curs.execute(_sql)
###提交
self.conn.commit()
#log.info('%s update ' % _sql)
except:
self.conn.rollback()
log.error('%s update error' % _sql)
raise ValueError('update error %s' % _sql)
return resNum
###删除
def delete(self, table, condition):
_sql = 'DELETE FROM `%s` WHERE %s' % (table, condition)
###执行
resNum = 0
try:
resNum = self.curs.execute(_sql)
###提交
self.conn.commit()
#log.info('%s delete ' % _sql)
except:
self.conn.rollback()
log.error('%s delete error' % _sql)
raise ValueError('112,delete error %s' % _sql)
return resNum
###直接给修改语句执行
def change(self, sql, many=False):
###过滤unknow table的warning
warnings.filterwarnings('ignore')
resNum = 0
if many:
try:
###多条同时插入
resNum = self.curs.executemany(sql, many)
self.conn.commit()
#log.info('%s exec ' % sql)
except:
self.conn.rollback()
log.error('%s exec error' % sql)
raise ValueError('exec error %s' % sql)
else:
try:
resNum = self.curs.execute(sql)
###提交
self.conn.commit()
#log.info('%s exec ' % sql)
except:
self.conn.rollback()
log.error('%s exec error' % sql)
raise ValueError('112,exec error %s' % sql)
return resNum
###查询
def query(self, sql):
res = ''
try:
self.curs.execute(sql)
res = self.curs.fetchall()
#log.info('%s query ' % sql)
except:
log.error('%s query error' % sql)
# raise ValueError('query error %s'% sql)
return res
if __name__ == "__main__":
args = dict()
args['host'] = '172.16.17.164'
args['user'] = 'miles'
args['pswd'] = 'aspect'
args['db'] = 'test'
sql_sel = "select * from bigdata_host limit 5"
m = MysqlBase(**args)
data = m.query(sql=sql_sel)
m.insert('bigdata_host', {'hostname': 'ccc', 'remark': 'ddd', 'up_addr_p': '192.168.136.41', 'states': '4',
'enter_time': '2017-03-13'})
m.delete('bigdata_host', 'hostname="ccc"')

25
ops/mylog.py Normal file
View File

@ -0,0 +1,25 @@
from __future__ import absolute_import
import logging
def define_logger(filename="/data/logs/aa.log", debug=True):
logger = logging.getLogger("")
if debug == True:
logger.setLevel(logging.INFO)
else:
logger.setLevel(logging.ERROR)
# 设置输出格式
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# 设置日志文件处理器
fh = logging.FileHandler(filename)
fh.setFormatter(formatter) # 为这个处理器添加格式
# 设置屏幕stdout输出处理器
formatter_stdout = logging.Formatter('%(name)s - %(levelname)s - %(message)s')
sh = logging.StreamHandler(stream=None)
sh.setFormatter(formatter_stdout)
# 把处理器加到logger上
logger.addHandler(fh) # logger.addHandler(sh)

54
web.py Normal file
View File

@ -0,0 +1,54 @@
# -*- coding: utf-8 -*-
# 推广系统对外接口，提供与客户端之间的广告信息接口及每分钟执行一次的缓存变更操作
# http://ad.kingsome.cn/webapp/index.php?c=Ops&a=getAdList&body={"gameid":1004,"locationid":1001}
# http://ad.kingsome.cn/webapp/index.php?c=Ops&a=upAdRecording&adid=1002
# python ad_interface_tornado.py --port=6013
import tornado.ioloop
import tornado.web
import tornado.options
from tornado import gen
import json
import datetime
from ops.mylog import define_logger
import logging
from config.config import *
import pdb
define_logger("/data/logs/ops/ops_interface.log")
log = logging.getLogger(__name__)
tornado.options.define("port", default=interface_port, type=int, help="run server on the given port.")
class DispatchHandler(tornado.web.RequestHandler):
@gen.coroutine
def get(self):
if self.get_query_argument('c') == 'Ops' and self.get_query_argument('a') == 'selfChecking':
self._selfCheckingHandler()
elif self.get_query_argument('c') == 'Ops' and self.get_query_argument('a') == 'getAdList':
yield self._selfGetAdList()
else:
self.write("pls check args!")
@gen.coroutine
def post(self):
if self.get_query_argument('c') == 'Ops' and self.get_query_argument('a') == 'upJumpRecording':
self._selfupJumpRecording()
def make_app():
return tornado.web.Application([(r"/webapp/index[\.]php", DispatchHandler)])
if __name__ == "__main__":
print('start!')
tornado.options.parse_command_line()
app = make_app()
app.listen(tornado.options.options.port)
tornado.ioloop.IOLoop.current().start()