# python3数据处理脚本模板 {#python3数据处理脚本模板}

本文给出一些数据处理脚本的模板。平时在我们上线新的功能时，可能会涉及到处理历史数据，那么一般情况下，建议使用python脚本来处理，原因是linux服务器默认内置python环境，不需要为处理脚本额外搭建环境，另外运维人员也比较熟悉python脚本，所以运维人员也可以把控到风险。向运维提交的数据处理脚本，优先考虑sql语句，其次考虑python脚本。

# 1. 示例1: 生成sql插入语句 {#_1-示例1-生成sql插入语句}

本示例实现的数据处理需求：将已关注公众号、且没有入库的用户，保存到数据库中(实际上是生成insert语句)。

本示例用到的知识点：

• 数据库读取
  借助pymysql。
• 请求接口
  借助requests，可以自动将接口返回的数据转换为json。
• snowflake
  生成snowflake ID。
• 日志打印
  封装了一个通用的日志记录器。
• 读取命令行参数
• 变量值注入到字符串
  借助python3的字符串格式化语法：f''。

该示例之所以生成了sql语句，而没有直接插入到数据库中。原因有3点：

• 给sql的方式沟通更简单高效，也利于一旦出错时的sql回滚

• 运行该脚本，还需要启动一个snowflake服务，运维还得多一步操作

  This is a sample Python script.

  Press ⌃R to execute it or replace it with your code.

  Press Double ⇧ to search everywhere for classes, files, tool windows, actions, and settings.

  import datetime import logging import sys

  import requests import pymysql import snowflake.client

  全局变量

  logger: logging.Logger = None mysql_connect: pymysql.connections.Connection = None mysql_host = None mysql_user = None mysql_pwd = None mysql_port = None mysql_db = None

  初始化日志记录器

  def init_logger(): global logger logger = logging.getLogger("data_process") logger.setLevel(logging.DEBUG) # 建立一个filehandler来把日志记录在文件里，级别为debug以上 fh = logging.FileHandler("data_process.log") fh.setLevel(logging.DEBUG) # 建立一个streamhandler来把日志打在CMD窗口上，级别为debug以上 ch = logging.StreamHandler() ch.setLevel(logging.DEBUG) # 设置日志格式 formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s") ch.setFormatter(formatter) fh.setFormatter(formatter) # 将相应的handler添加在logger对象中 logger.addHandler(ch) logger.addHandler(fh)

  初始化mysql连接

  def init_mysql_connect(): global mysql_host, mysql_user, mysql_pwd, mysql_port, mysql_db, mysql_connect

    mysql_connect = pymysql.connect(host=mysql_host, user=mysql_user, password=mysql_pwd, port=mysql_port,
                                    database=mysql_db)
  

  请求微信服务号的关注用户的openid列表

  def request_fuwuhao_openids(access_token): api = f'https://api.weixin.qq.com/cgi-bin/user/get?access_token={access_token}' wx_openids = requests.get(api).json()

    logger.info("请求微信服务号的关注用户的openid列表, 总数:%s", wx_openids['total'])
    return wx_openids["data"]["openid"]
  

  请求服务号的用户列表

  def request_new_fuwuhao_users(): global mysql_connect, access_token

    weixin_users = []
  
    openids = request_fuwuhao_openids(access_token)
    index = 0
    for openid in openids:
        logger.info("请求服务号的用户列表, 当前进度:%s, 总数: %s", index, len(openids))
        index = index + 1
  
        # 若当前openid在数据库中不存在，则补充到weixin_users数组
        if get_fuwuhao_user_from_db(openid) is None:
            logger.info("请求服务号的用户列表, 当前openid在数据库中不存在: %s", openid)
            weixin_user = request_weixin_user(openid)
            weixin_users.append(weixin_user)
        else:
            logging.info("请求服务号的用户列表, 当前openid在数据库中已存在: %s", openid)
  
    logger.info("请求服务号的用户列表, 完成, total: %s, count_will_process:%s", len(openids), len(weixin_users))
    return weixin_users
  

  请求某微信用户的信息

  def request_weixin_user(openid): global access_token

    api = f'https://api.weixin.qq.com/cgi-bin/user/info?lang=zh_CN&access_token={access_token}&openid={openid}'
    wx_user = requests.get(api).json()
    logger.info("请求某微信用户的信息: %s", wx_user)
    return wx_user
  

  从数据库中查询某服务号用户

  def get_fuwuhao_user_from_db(openid): global mysql_connect cursor = mysql_connect.cursor()

    sql = f"select 1 from fuwuhao_users where openid = '{openid}'"
    cursor.execute(sql)
    return cursor.fetchone()
  

  生成插入sql

  def generate_insert_sql(): global mysql_connect, access_token

    # 请求服务号关注用户列表
    wx_users = request_new_fuwuhao_users()
  
    # 生成sql
    sql = "INSERT INTO fuwuhao_users (id, is_subscribe, openid, lang, subscribe_time, unionid, remark, groupid, tagids, subscribe_scene, qr_scene, qr_scene_str, created_at, updated_at, deleted_at) VALUES "
    cur_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    index = 0
    for elem in wx_users:
        sql = sql + f"({generate_snowflake_id()}, '{elem['subscribe']}', '{elem['openid']}', '{elem['language']}', {elem['subscribe_time']}, '{elem['unionid']}', '{elem['remark']}', {elem['groupid']}, '{elem['tagid_list']}', '{elem['subscribe_scene']}', {elem['qr_scene']}, '{elem['qr_scene_str']}', '{cur_time}', '{cur_time}', null)"
  
        if index != len(wx_users) - 1:
            sql = sql + ","
  
        index = index + 1
  
    return sql
  

  生成一个snowflake id

  python3生成snowflake的步骤：

  1. pip3 install pysnowflake

  2. 运行snowflake_start_server

  3. 代码中调用snowflake.client.get_guid()获取一个snowflake ID

  def generate_snowflake_id(): # 配置snowflake服务的主机配置，若不配置, 则默认使用localhost:8910 snowflake_server_host = 'localhost' snowflake_server_port = 8910 snowflake.client.setup(snowflake_server_host, snowflake_server_port)

    return snowflake.client.get_guid()
  

  Press the green button in the gutter to run the script.

  if name == 'main': # 从命令行读取参数 # 程序运行示例：python3 ./main.py weixin_api_access_token database_1 127.0.0.1 root 123456 3306 cmd_args = sys.argv access_token = cmd_args[1] mysql_db = cmd_args[2] mysql_host = cmd_args[3] mysql_user = cmd_args[4] mysql_pwd = cmd_args[5] mysql_port = int(cmd_args[6])

    # 初始化
    init_logger()
    init_mysql_connect()
  
    # 生成数据库表插入语句
    sql = generate_insert_sql()
    logger.info("main, 生成sql语句完成: %s", sql)