<file_sep>#coding:utf8

"""
Author : zhangwenqi
Description : 读取excel内容
CreateTime : 2019-08-05
"""
import openpyxl

class ObtainEexcel():
    def __init__(self,filename,sheetname):
        self.filename = filename
        self.handxl = openpyxl.load_workbook(filename)
        self.sheet = self.handxl.get_sheet_by_name(sheetname)
        self.row_no = self.sheet.max_row
        self.column_no = self.sheet.max_column

    """获取所有的sheet名称"""
    def obtain_sheets_name(self):
        return self.handxl.get_sheet_names()

    """获取表格内容"""
    def obtain_table_value(self,row,column):
        return self.sheet.cell(row=row,column=column).value

    """写入表格内容"""
    def set_table_value(self, row, column,value):
        self.sheet.cell(row=row, column=column).value = value

    """获取所有行数据"""
    def obtain_table_rows_value(self):
        list_rows_value = []
        for i in self.sheet.iter_rows():
            list_rows_value.append([j.value for j in i])
        return list_rows_value

    """获取所有列数据"""
    def obtain_table_columns_value(self):
        list_columns_value = []
        for i in self.sheet.iter_cols():
            list_columns_value.append([j.value for j in i])
        return list_columns_value

    """通过表名标题去找对应数据"""
    def obtain_position_by_table(self,table_name):
        tables = self.sheet.iter_rows().__next__()
        return [i.value for i in tables].index(table_name)+1

    def obtain_values_by_position(self,table_name,target_value):
        position = self.obtain_position_by_table(table_name)
        list_values = self.obtain_table_rows_value()
        result_values = []
        for i in list_values:
            if i[position-1] == target_value:
                result_values.append(i)
        return result_values

    def table_values_sort(self,table_name,values):
        position = self.obtain_position_by_table(table_name)
        return sorted(values,key=lambda x:x[position-1])


    """保存excel"""
    def save(self):
        self.handxl.save(self.filename)
        self.handxl.close()<file_sep>from flask import Blueprint
target_cycle_summary = Blueprint('target_cycle_summary', __name__)

# 查询目标库
@target_cycle_summary.route('/query_name/<user_name>', methods=['GET'])
def query_by_name(user_name):
    return "hello1"<file_sep>#coding:utf8
"""
Author : zhangwenqi
Description : http请求发送工具
CreateTime : 2019-08-05
"""
import requests
from conf import test_jar_conf
from common.result import ResultMessage
class ObtainRequests():
    def __init__(self,url,method,headers=None,query=None,body=None,body_type=None):
        self.url = url
        self.method = method
        self.headers = headers
        self.query = query
        self.body = body
        self.body_type = body_type

    """
        Description : 发请求实体
        CreateTime : 2019-08-05
    """
    def obtain_requests(self):
        method = self.method.upper()
        if self.body_type is not None and isinstance(self.body_type,str):
            self.body_type = self.body_type.upper()
        code = self._check_requests_parame(self.url,self.method,self.headers,self.query,self.body,self.body_type)
        respons = None
        result_message = ResultMessage()
        if code != test_jar_conf.API_REQUESTS_SUCCESS_CODE:
            result_message.errcode = code
            return result_message.result()
        if method == "GET":
            if self.headers is not None:
                if self.query is not None:
                    respons = requests.get(self.url,headers=self.headers,params=self.query)
            else:
                if self.query is not None:
                    respons = requests.get(self.url,params=self.query)
                else:
                    respons = requests.get(self.url)
        if method == "POST":
            if self.headers is not None:
                if self.query is not None:
                    if self.body is not None:
                        if self.body_type == "JSON":
                            #body = json.dumps(body)
                            respons = requests.post(self.url,headers=self.headers,params=self.query,json=self.body)
                        elif self.body_type == "FORM":
                            respons = requests.post(self.url,headers=self.headers,params=self.query,data=self.body)
                    else:
                        respons = requests.post(self.url, headers=self.headers, params=self.query)
                else:
                    if self.body is not None:
                        if self.body_type == "JSON":
                            #body = json.dumps(body)
                            respons = requests.post(self.url,headers=self.headers,json=self.body)
                        elif self.body_type == "FORM":
                            respons = requests.post(self.url,headers=self.headers,data=self.body)
                    else:
                        respons = requests.post(self.url, headers=self.headers)
            else:
                if self.query is not None:
                    if self.body is not None:
                        if self.body_type == "JSON":
                            # body = json.dumps(body)
                            respons = requests.post(self.url, params=self.query, json=self.body)
                        elif self.body_type == "FORM":
                            respons = requests.post(self.url, params=self.query, data=self.body)
                    else:
                        respons = requests.post(self.url ,params=self.query)
                else:
                    if self.body is not None:
                        if self.body_type == "JSON":
                            # body = json.dumps(body)
                            respons = requests.post(self.url, json=self.body)
                        elif self.body_type == "FORM":
                            respons = requests.post(self.url, data=self.body)
                    else:
                        respons = requests.post(self.url)

        result_message.data = respons
        result_message.errcode = code
        return result_message

    """
    Description : 校验请求参数准确性方法
    CreateTime : 2019-08-05
    """
    @staticmethod
    def _check_requests_parame(url,method,headers=None,query=None,body=None,body_type=None):
        if not isinstance(url,str):
            return test_jar_conf.API_REQUESTS_URL_TYPE_ERROR_CODE
        if not url.startswith("http://") and not url.startswith("https://"):
            return test_jar_conf.API_REQUESTS_URL_TYPE_ERROR_CODE
        if not isinstance(method,str):
            return test_jar_conf.API_REQUESTS_METHOD_CONTENT_ERROR_CODE
        if method not in test_jar_conf.API_REQUESTS_METHOD_LIST:
            return test_jar_conf.API_REQUESTS_METHOD_CONTENT_ERROR_CODE
        if headers is not None:
            if isinstance(headers, str):
                try:
                    headers = eval(headers)
                    if not isinstance(headers,dict):
                        return test_jar_conf.API_REQUESTS_HEADER_CONTENT_ERROR_CODE
                except:
                    return test_jar_conf.API_REQUESTS_HEADER_CONTENT_ERROR_CODE
            elif not isinstance(headers, dict):
                return test_jar_conf.API_REQUESTS_HEADER_CONTENT_ERROR_CODE
        if query is not None:
            if isinstance(query, str):
                try:
                    query = eval(query)
                    if not isinstance(query, dict):
                        return test_jar_conf.API_REQUESTS_QUERY_CONTENT_ERROR_CODE
                except:
                    return test_jar_conf.API_REQUESTS_QUERY_CONTENT_ERROR_CODE
            elif not isinstance(query, dict):
                return test_jar_conf.API_REQUESTS_QUERY_CONTENT_ERROR_CODE
        if body is not None:
            if isinstance(body, str):
                try:
                    body = eval(body)
                    if not isinstance(body, dict):
                        return test_jar_conf.API_REQUESTS_BODY_CONTENT_ERROR_CODE
                except:
                    return test_jar_conf.API_REQUESTS_BODY_CONTENT_ERROR_CODE
            elif not isinstance(body, dict):
                return test_jar_conf.API_REQUESTS_BODY_CONTENT_ERROR_CODE
        if not body_type in test_jar_conf.API_REQUESTS_BODY_TYPE_LIST:
            return test_jar_conf.API_REQUESTS_BODY_TYPE_ERROR_CODE
        return test_jar_conf.API_REQUESTS_SUCCESS_CODE

if __name__ == "__main__":
    o = ObtainRequests("http://www.didi.com","GET")
    print(o.obtain_requests().data.text)<file_sep>#coding:utf8
"""
Author : zhangwenqi
Description : 定义统一返回结果
CreateTime : 2019-08-21
"""

import json

class ResultMessage:
    def __init__(self):
        #响应状态码
        self.__errcode = None

        #响应结果
        self.__message = None

        #传输数据
        self.__data = None

    @property
    def errcode(self):
        return self.__errcode

    @property
    def message(self):
        return self.__message

    @property
    def data(self):
        return self.__data

    @errcode.setter
    def errcode(self,errcode):
        if isinstance(errcode,int):
            self.__errcode = errcode
        else:
            raise TypeError("需要传入整型")

    @message.setter
    def message(self,message):
        if isinstance(message,str):
            self.__message = message
        else:
            raise TypeError("需要传入字符串")

    @data.setter
    def data(self,data):
        self.__data = data

    def result(self):
        dict_result = {"errcode":self.__errcode,"message":self.__message,"data":self.__data}
        return json.dumps(dict_result)<file_sep># -*- coding: utf-8 -*-
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.automap import automap_base

class BaseDao(object):

    #数据库表对象
    __base_class = None
    #数据库表名
    _table_name = None
    #数据库配置
    _db_conf = None

    @staticmethod
    def init(app):
        print(app)
        BaseDao._db_conf = app

    def __init__(self,encoding='utf-8', echo=False):
        self._engine = create_engine(BaseDao._db_conf.config.get("SQLALCHEMY_DATABASE_URI"), encoding=encoding, echo=echo)
        session_class = sessionmaker(self._engine)
        self._session = session_class()

    def _table_query(self, table_name):
        """ 避免重复加载，同时只是用execute时，不加载以下内容 """
        if self.__base_class is None:
            self.__base_class = automap_base()
            self.__base_class.prepare(self._engine, reflect=True)
        if hasattr(self.__base_class.classes, table_name):
            table = getattr(self.__base_class.classes, table_name)
            return self._session.query(table)
        else:
            raise KeyError(f"未搜寻到表: {table_name}")

    def execute_sql(self, sql_str, ft = {}):
        """
        执行sql语句，区分查询跟修改操作
        cursor支持fetchall(),fetchone(),fetchmany(),rowcount
        """
        sql_str = sql_str.format(**ft).strip().replace("\n", "").replace("\t", "")
        if sql_str.startswith("select"):
            cursor = self._engine.execute(sql_str)
            return cursor
        try:
            affect_row = self._engine.execute(sql_str).rowcount
            self._session.commit()
            return affect_row
        except Exception as e:
            self._session.rollback()
            self.close()
            raise e

    def execute_many_sql(self, many_sql, ft,separator=";"):
        """ 执行多条sql, sql语句之间使用；号分割，方便执行多条修改语句 """
        sql_list = many_sql.format(**ft).strip().replace("\n", "").replace("\t", "").split(separator)
        sql_list = [sql for sql in sql_list if sql.strip() != ""]
        for sql_str in sql_list:
            self.execute_sql(sql_str,ft)

    def delete(self, ft=None):
        """ 表名 + dict(key=word), 根据ft条件删除数据 """
        ft = {} if ft is None else ft
        delete_row = self._table_query(self._table_name).filter_by(**ft).delete()
        self._session.commit()
        return delete_row

    def first(self, ft=None):
        """ 一行数据 """
        ft = {} if ft is None else ft
        first_row = self._table_query(self._table_name).filter_by(**ft).first()
        return first_row

    def all(self, ft=None):
        """ 所有数据 """
        ft = {} if ft is None else ft
        all_row = self._table_query(self._table_name).filter_by(**ft).all()
        return all_row

    def count(self, ft=None):
        """ 行数 """
        ft = {} if ft is None else ft
        rowcount = self._table_query(self._table_name).filter_by(**ft).count()
        return rowcount

    def close(self):
        self._session.close()

    # def param_check(self,data,type = BaseDao._CHECKOUT_TYPE_TABLES):
    #     if type == BaseDao._CHECKOUT_TYPE_TABLES:
    #         for j in data:
    #             for i in self.__user_dict.keys():
    #                 if i not in j.keys():
    #                     return False
    #     elif type == BaseDao._CHECKOUT_TYPE_TABLE:
    #         for i in data.keys():
    #             if i not in self.__user_dict.keys():
    #                 return False
    #     return True

    # #校验单条数据
    # _CHECKOUT_TYPE_TABLE = "table"
    # #校验多条数据
    # _CHECKOUT_TYPE_TABLES = "tables"
    #设置表对象
    _SET_TYPE_TABLE = "table"
    #设置现成数据
    _SET_TYPE_TABLE_DATA = "table_data"

if __name__ == '__main__':
    pass



<file_sep>"""

"""

from dao.system_user_info_dao import UserInfo
from common.result import ResultMessage
from conf import test_jar_conf
import json

class SystemUserInfoService:
    result_message = ResultMessage()

    def __init__(self):
        self.user_info = UserInfo()

    def find_user_info_by_id(self,id):
        result = self.user_info.find_by_id(id)
        return result

    def update_user_info(self,user_info):
        result = None
        if not user_info.get("id"):
            result = self.user_info.insert_user(user_info)
            return result

        update_user_info = self.user_info.find_by_id(user_info["id"])
        if update_user_info:
            result = self.user_info.update_user_by_id(user_info)
        return result

    def delete_user_info(self,id):
        user_info = {"delete_status":1,"id":id}
        result = self.user_info.update_user_by_id(user_info)
        return result



if __name__ == "__main__":
    pass
    # SystemUserInfoService.find_user_info_by_id(1)<file_sep>from flask import Blueprint,request
from service.system_user_info_service import SystemUserInfoService
import json
from common.result import ResultMessage
from conf import test_jar_conf
system_user_info = Blueprint('system_user_info', __name__)

@system_user_info.route('/get/<id>', methods=['GET'])
def get(id):
    result_message = ResultMessage()
    system_user_info_service = SystemUserInfoService()
    result = system_user_info_service.find_user_info_by_id(id)

    result_message.data = result
    result_message.errcode = test_jar_conf.SUCCESS_CODE
    return result_message.result()

@system_user_info.route('/update', methods=['POST'])
def update():
    result_message = ResultMessage()
    user_info = json.loads(request.get_data(as_text=True))

    system_user_info_service = SystemUserInfoService()
    result = system_user_info_service.update_user_info(user_info)

    result_message.data = result
    result_message.errcode = test_jar_conf.SUCCESS_CODE
    return result_message.result()

@system_user_info.route('/delete/<id>', methods=['GET'])
def delete(id):
    result_message = ResultMessage()
    system_user_info_service = SystemUserInfoService()
    result = system_user_info_service.delete_user_info(id)

    result_message.data = result
    result_message.errcode = test_jar_conf.SUCCESS_CODE
    return result_message.result()
<file_sep>from dao.base_dao import BaseDao
import copy

class UserInfo(BaseDao):

    __user_dict = {"id":None,"user_name":None,"group_id":None,"email":None,"password":<PASSWORD>,"delete_status":None}
    _table_name = "system_user"

    def find_by_id(self,id = None):
        user_data = BaseDao.first(self,{"id":id,"delete_status":0})
        return self.__set_user(user_data)


    def find_by_condition(self,ft = None):
        user_list = []
        user_data = BaseDao.all(self,ft)
        for i in user_data:
            user = self.__set_user(i)
            user_list.append(copy.copy(user))
        return user_list

    def all(self):
        user_list = []
        user_data = BaseDao.all(self)
        for i in user_data:
            user = self.__set_user(i)
            user_list.append(copy.copy(user))
        return user_list

    def insert_users(self,user_list):
        for user_info in user_list:
            self.insert_user(user_info)

    def insert_user(self,user_info):
        sql_str = "INSERT INTO system_user (user_name,group_id,email,password) VALUES ('{user_name}','{group_id}','{email}','{password}');"
        return self.execute_sql(sql_str,user_info)

    def update_user_by_id(self,modify):
        condition = {"id":modify.pop("id")}
        return self.update_users(modify,condition)

    def update_users(self,modify,condition):
        sql_str = "UPDATE system_user SET "
        for i in modify.keys():
            sql_str += "%s = '%s'," % (i,modify[i])
        sql_str = sql_str[:-1] + " where "
        for i in condition.keys():
            sql_str += "%s = '%s' AND" % (i,condition[i])
        sql_str = sql_str[:-3]+";"
        print(sql_str)
        return self.execute_sql(sql_str)

    def __set_user(self,user_data,type=BaseDao._SET_TYPE_TABLE):
        if not user_data:
            return
        if type == BaseDao._SET_TYPE_TABLE:
            UserInfo.__user_dict["id"] = user_data.id
            UserInfo.__user_dict["user_name"] = user_data.user_name
            UserInfo.__user_dict["group_id"] = user_data.group_id
            UserInfo.__user_dict["email"] = user_data.email
            UserInfo.__user_dict["password"] = <PASSWORD>
            UserInfo.__user_dict["delete_status"] = user_data.delete_status
            return UserInfo.__user_dict
        else:
            UserInfo.__user_dict["user_name"] = user_data.user_name
            UserInfo.__user_dict["group_id"] = user_data.group_id
            UserInfo.__user_dict["email"] = user_data.email
            UserInfo.__user_dict["password"] = <PASSWORD>
            UserInfo.__user_dict["delete_status"] = user_data.delete_status
            return UserInfo.__user_dict




if __name__ == "__main__":
    m = "mysql+pymysql://root:123@localhost:3306/atlantis?charset=utf8mb4"
    UserInfo.db_conf = m
    u = UserInfo()
    print(u.find_by_id(1))
    u.update_users({"name":"ztb1","email":"test1"},{"id":2})
    # print(u.find_by_condition({"name":"2"}))
    # u.insert_users([{"id1":2,"name":"2","business_id":2,"email":"2","pwd":1,"status":1,"company_name":1},{"id":3,"name":"3","business_id":'3',"email":'3',"pwd":'3',"status":'3',"company_name":'3'}])
    # print(l)
<file_sep>from flask import Flask
from dao.base_dao import BaseDao
from conf import config
from controller.system_user_info import system_user_info
from controller.target_cycle_summary import target_cycle_summary
from controller.obtain_target_summary import obtain_target_summary

app = Flask(__name__)
# 引入配置文件中的相关配置
app.config.from_object(config.config["development"])
# 配置db
BaseDao.init(app)
# 注册蓝图，并指定其对应的前缀（url_prefix）
app.register_blueprint(system_user_info, url_prefix="/system_user_info")
app.register_blueprint(target_cycle_summary, url_prefix="/target_cycle_summary")
app.register_blueprint(obtain_target_summary, url_prefix="/obtain_target_summary")



@app.route('/')
def hello_world():
    return 'Hello World!'


if __name__ == '__main__':
    app.run(host="127.0.0.1", port=5000, debug=True)
<file_sep>import openpyxl

def func(s):
    for i in s:
        print("%s:%s" % (i,s[i]))


if __name__ == "__main__":
    s = {"sign":"44549876042bc0cce5f57fd1432e6b92","company_id":"8968336369328084993","out_order_id":"21666097","callback_status":"10001","order_id":"611986578947567258","order_source":"1","new_order":"{\"order_id\":\"14110766\"}","msg":"","test_function":"OrderStatusCallbackTest_testAppRestartOrder#"}

    func(s)
    # file = r"/Users/didi/Desktop/111.xlsx"
    # wb = openpyxl.load_workbook(file)
    # wb.create_sheet("3")
    # d = {}
    # st = wb["1"]
    # n = st.max_row
    # for i in range(2,n+1):
    #     v = (lambda x: x.strip() if x else x)(st["A%s" % i].value)
    #     if v:
    #         if v not in d:
    #             d[v] = {"j1":st["B%s" % i].value,"b":"1","j2":"","j3":""}
    #         else:
    #             if d[v]["j1"] != st["B%s" % i].value:
    #                 d[v]["j3"] = st["B%s" % i].value
    #                 d[v]["b"] = "2"
    # st = wb["2"]
    # n = st.max_row
    # for i in range(2,n+1):
    #     v = (lambda x: x.strip() if x else x)(st["A%s" % i].value)
    #     if v:
    #         if v not in d:
    #             d[v] = {"j1":st["B%s" % i].value,"b":"1","j2":"","j3":""}
    #         else:
    #             if d[v]["j1"] != st["B%s" % i].value:
    #                 d[v]["j2"] = st["B%s" % i].value
    #                 d[v]["b"] = "2"
    # st = wb["3"]
    # n = 2
    # for i in d:
    #     st["A%s" % n] = i
    #     st["B%s" % n] = d[i]["j1"]
    #     st["C%s" % n] = d[i]["j2"]
    #     st["D%s" % n] = d[i]["j3"]
    #     st["E%s" % n] = d[i]["b"]
    #     n += 1
    # wb.save(file)<file_sep>from flask import Blueprint
obtain_target_summary = Blueprint('obtain_target_summary', __name__)

# 查询目标库
@obtain_target_summary.route('/query_name/<name>', methods=['GET'])
def query_by_name(name):
    return "hello3"+name<file_sep>#coding:utf8
"""
Author : zhangwenqi
Description : testjJar全量配置
CreateTime : 2019-08-05
"""
#外部域名





#成功状态码
SUCCESS_CODE = 0
#参数类型错误状态码
PARAMETER_TYPE_ERROR = 1001
#参数缺失错误
PARAMETER_MISSING_ERROR = 1002
#请求失败
REQUEST_FAILURE_ERROR = 1010

#http请求发送工具配置
#成功
API_REQUESTS_SUCCESS_CODE = 2000
#请求url格式错误
API_REQUESTS_URL_TYPE_ERROR_CODE = 2001
#方法内容错误
API_REQUESTS_METHOD_CONTENT_ERROR_CODE = 2002
#主体内容错误
API_REQUESTS_BODY_CONTENT_ERROR_CODE = 2003
#请求头错误
API_REQUESTS_HEADER_CONTENT_ERROR_CODE = 2004
#主体格式错误
API_REQUESTS_BODY_TYPE_ERROR_CODE = 2005
#QUERY格式错误
API_REQUESTS_QUERY_CONTENT_ERROR_CODE = 2006

#请求主体类型
API_REQUESTS_BODY_TYPE_LIST = ["FORM","JSON",None]

#HTTP请求方法
API_REQUESTS_METHOD_LIST = ["POST","GET","DELETE","PUT","HEAD"]<file_sep>import os

BASEDIR = os.path.abspath(os.path.join(os.path.dirname("__file__"),os.path.pardir))


class Config:
    """base config"""
    # SECRET_KEY = os.environ.get('SECRET_KEY') or 'secret key'
    # SQLALCHEMY_COMMIT_ON_TEARDOWN = True
    # SQLALCHEMY_TRACK_MODIFICATIONS = True
    # FLASKY_ADMIN = os.environ.get('FLASKY_ADMIN')

    FLASKY_MAIL_SENDED = '<EMAIL>'  # 发件人地址
    FLASKY_MAIL_SUBJECT_PREFIX = '[Flasky]'  # 邮件主题前缀


class ProductionConfig(Config):
    """运行环境配置"""
    DEBUG = True
    DIALCT = 'mysql'
    DRIVER = "pymysql"
    USERNAME = 'root'
    PASSWORD = '<PASSWORD>!@#'
    HOST = 'localhost'
    PORT = '3306'
    DBNAME = 'practice'
    SQLALCHEMY_DATABASE_URI = '{}://{}:{}@{}:{}/{}?charset=utf8mb4'.format(DIALCT, USERNAME, PASSWORD, HOST,PORT, DBNAME)
    SQLALCHEMY_TRACK_MODIFICATIONS = True  # 没有此配置会导致警告



class DevelopmentConfig(Config):
    """运行环境配置"""
    DEBUG = True
    DIALCT = 'mysql'
    DRIVER = "pymysql"
    USERNAME = 'root'
    PASSWORD = '<PASSWORD>'
    HOST = 'localhost'
    PORT = '3306'
    DBNAME = 'practice'
    SQLALCHEMY_DATABASE_URI = '{}://{}:{}@{}:{}/{}?charset=utf8mb4'.format(DIALCT, USERNAME, PASSWORD, HOST, PORT,
                                                                           DBNAME)
    SQLALCHEMY_TRACK_MODIFICATIONS = True  # 没有此配置会导致警告


config = {
    'development': DevelopmentConfig,
    'testing': ProductionConfig,
    'production': ProductionConfig,
    'default': DevelopmentConfig
}

if __name__ == "__main__":
    print(ProductionConfig.SQLALCHEMY_DATABASE_URI)