python-cookbook

概述

分析日志相关的脚本用的比较多。写一篇文章，收集python常用的一些方法。

环境搭建

python

Python 3.9.0

实际需要掌握的技巧

正则表达式

# [18:57:52] [INF] GiftItemLog buy rid: 15909, ID: 601, Type: 6, Rmb: 30
re_buy = re.compile("GiftItemLog buy rid: ([0-9_]+), ID: ([0-9_]+), Type: ([0-9_]+), Rmb: ([0-9_]+)")

按照utf-8来读写文件,按照行读取文件

c_fd = codecs.open('./insert_create.sql', 'w', encoding='utf-8')
with codecs.open('./2020-04-22.log', 'r', encoding='utf-8') as fd:
    while True:
        line = fd.readline()
        if not line: break
c_fd.close()

写入excel中

日志如何编写

官网日志文档

import logging
logging.warning('Watch out!')  # will print a message to the console
logging.info('I told you so')  # will not print anything

# 设置输出文件
import logging
logging.basicConfig(filename='example.log', encoding='utf-8', level=logging.DEBUG)
logging.debug('This message should go to the log file')
logging.info('So should this')
logging.warning('And this, too')
logging.error('And non-ASCII stuff, too, like Øresund and Malmö')

# 设置输出的日期格式
import logging
logging.basicConfig(level=logging.INFO,format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',datefmt='%a, %d %b %Y %H:%M:%S')
logging.warning('is when this event was logged.')

支持循环日志

RotatingFileHandler

logging-handlers

import logging
from logging.handlers import RotatingFileHandler

logger = logging.getLogger('mylogger')
logger.setLevel(logging.INFO)

handler = RotatingFileHandler(filename='test.log', maxBytes=6, backupCount=3)

logger.addHandler(handler)

for i in range(10):
    logger.info("Hello%d" % i)

int转换str填充0

参考int转换str填充0

str(1).zfill(2)

Pandas

Pandas 是一个流行的 Python 数据分析库，它提供了强大的数据结构和数据分析工具。Pandas 的主要特点包括：

1. 易于使用的数据结构和数据分析工具：Pandas 提供了一系列易于使用的数据结构，如 Series、DataFrame 和 GroupBy，以及强大的数据分析工具，如聚合函数和统计函数。
2. 高效的数据处理能力：Pandas 使用高性能的 Cython 库进行计算，可以快速处理大规模的数据。
3. 灵活的数据操作：Pandas 支持各种数据操作，如添加或删除列、合并数据、筛选数据等。
4. 丰富的数据导入和导出功能：Pandas 支持从各种数据源导入数据，如 CSV、Excel、SQL 数据库等，并可以将数据导出到这些格式。
5. 与其他 Python 库集成：Pandas 可以与其他流行的 Python 库集成，如 NumPy、Matplotlib 和 Scikit-learn。

以下是一些 Pandas 的常用功能：

1. 读取和写入数据：使用 Pandas 可以轻松地读取和写入 CSV、Excel、SQL 数据库等格式的数据。
2. 数据清洗和处理：Pandas 提供了各种数据清洗和处理功能，如缺失值处理、重复值处理、数据类型转换等。
3. 数据聚合和分析：Pandas 支持对数据进行聚合和分析，如计算平均值、总和、标准差等统计指标。
4. 数据可视化：Pandas 可以与 Matplotlib 库结合使用，实现数据可视化功能。
5. 数据处理和管理：Pandas 提供了强大的数据管理功能，如合并数据、分割数据、筛选数据等。

总之，Pandas 是一个功能强大的 Python 数据分析库，可以广泛应用于数据处理、分析和管理等方面。

def _get_db_connect(db_config, host=None):
    return pymysql.connect(host=host if host else db_config['host_ro'],
                           user=db_config['user'],
                           password=db_config['pwd'],
                           db=db_config['db'],
                           charset="utf8mb4",
                           cursorclass=pymysql.cursors.DictCursor)

Anaconda python版本管理

conda

ms_engine = create_engine("mysql+pymysql://root:123456@localhost:3306/test")
import pandas as pd
from sqlalchemy import create_engine
 
DATABASE = {
    '69': {
        'host': 'localhost',
        'user': 'root',
        'password': '123456',
        'database': 'test',
        'port': 3306
    }
}
 
# 连接哪台服务器
TB_CONNECT = "69"
DATABASE = DATABASE[TB_CONNECT]
 
def save_data_to_mysql(df, table_name):
    """
    将一个DataFrame保存至数据库
    
    参数说明:
        df: 一个DataFrame对象
        table_name: 需要存入数据库的表名,字符串类型
    """
    try:
        ms_engine = create_engine(
            'mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8'.format(
                DATABASE['user'], DATABASE['password'], DATABASE['host'],
                DATABASE['port'], DATABASE['database']))
 
        df.to_sql(table_name, ms_engine, if_exists='replace', index=False)
    except Exception as e:
        raise Exception("保存数据时发生错误：" + str(e))
    finally:
        ms_engine.dispose()
 
df_res = pd.DataFrame([[1,2,3,4],
                       [5,6,7,8],
                       [9,10,11,12]
                      ],columns=['A','B','C','D'])
 
# 将df_res数据存入到test数据库的df_result表中
save_data_to_mysql(df_res,'df_result')

pymysql 库的使用

python 定时执行库

import schedule  
import time  
import subprocess  
  
def run_script():  
    print("run_script")
    subprocess.call(["nohup", "/home/abel/crontab.sh"])  
  
# 设置定时任务，每周一、三、五的早上3点执行run_script函数  
schedule.every().monday.at("03:00").do(run_script)
schedule.every().wednesday.at("03:00").do(run_script)
schedule.every().friday.at("03:00").do(run_script)
  
while True:  
    schedule.run_pending()  
    time.sleep(1)

密码

import random
import string

def generate_password(length=16):
    characters = string.ascii_letters + string.digits
    password = ''.join(random.choice(characters) for _ in range(length))
    return password

# 生成密码并输出
generated_password = generate_password()
print("Generated Password:", generated_password)

实现将秒数转换成人类可读的时分秒格式

def seconds_to_hms(seconds):
    hours, remainder = divmod(seconds, 3600)
    minutes, seconds = divmod(remainder, 60)
    return "{:02}:{:02}:{:02}".format(int(hours), int(minutes), int(seconds))

# 示例
seconds = 3665
formatted_time = seconds_to_hms(seconds)
print(formatted_time)

要使用 Python 处理 block_scan.csv 文件并按照你的要求进行操作，可以使用 pandas 库。以下是完整的代码示例：

1. 按照第2列进行分组。
2. 在每个分组内按照第4列从小到大排序。
3. 找出第4列数字不连续的行并输出。

import pandas as pd

# 读取 CSV 文件
df = pd.read_csv('block_scan.csv')

# 按第2列分组
grouped = df.groupby(df.columns[1])

# 定义一个列表来存储排序后的分组数据
sorted_groups = []

# 定义查找不连续数字的函数
def find_discontinuities(group):
    sorted_group = group.sort_values(by=group.columns[3])
    sorted_groups.append(sorted_group)  # 将排序后的数据保存到列表
    sorted_group['diff'] = sorted_group[sorted_group.columns[3]].diff()
    discontinuities = sorted_group[sorted_group['diff'] != 1]
    return discontinuities[[group.columns[1], group.columns[3]]]

# 找出每个分组中第4列数字不连续的行
discontinuities = grouped.apply(find_discontinuities).reset_index(drop=True)

# 保存不连续的数字和对应的分组到 CSV 文件
discontinuities.to_csv('discontinuities.csv', index=False)

# 合并排序后的分组数据
sorted_df = pd.concat(sorted_groups).reset_index(drop=True)

# 删除辅助列 'diff'
sorted_df = sorted_df.drop(columns=['diff'])

# 保存排序后的结果到新的 CSV 文件
sorted_df.to_csv('sorted_block_scan.csv', index=False)

print("处理完成，已保存排序后的CSV文件和不连续的数字到discontinuities.csv文件中。")

代码解释：

1. 加载 CSV 文件：

   • 使用 pd.read_csv 加载 block_scan.csv 文件。

2. 分组：

   • 使用 groupby 按照 CSV 文件的第2列进行分组（通过列的索引访问）。

3. 排序和查找不连续行：

   • 定义一个函数 find_discontinuities，对每个分组按照第4列进行排序，并通过计算相邻行的差值来查找不连续的行。
   • 使用 diff 方法计算相邻行的差值，如果差值不为1，则认为该行是不连续的。

4. 应用函数：

   • 使用 apply 方法将 find_discontinuities 函数应用于每个分组，并将结果重置索引。

5. 删除辅助列：

   • 删除辅助列 diff，仅保留原始数据中的不连续行。

6. 输出结果：

   • 打印不连续的行。
   • 如果需要，可以将结果保存到新的 CSV 文件中。

确保 block_scan.csv 文件存在于当前工作目录下，或者提供文件的完整路径。运行该代码需要确保已安装 pandas 库，可以通过 pip install pandas 安装。

参考

• 官网日志文档
• pandas-pydata-user-guide