项目情况介绍： 基于Python 3.6.6 ,实现对nginx访问的日志分析代码，实现了对日志中code的占比统计和浏览器类型和访问情况统计 实现的代码段有： 1.编写窗户函数，实现在一定的时间内对数据进行分析 2.通过正则表达式对日志进行匹配，加载日志文件，提取出文本里每行的日志信息 3.编写消费端代码，即使得提取到的数据能够按照消费端的代码进行处理 4.消息分发代码实现，通过queue，将提取的的文本放到队列里，供消费端代码处理 项目代码如下
代码语言：javascript复制import random
import datetime
import time
from queue import Queue
import threading
import re
from pathlib import Path
from user_agents import parse
"""
这段代码，实现了再一段时间内获得数据，通过不同的handler（即消费端函数）
对获取到的同一份数据进行处理，主要是两段消费函数，网页返回的code的统计和浏览器的分析
这段代码，窗口函数中，data = src.get(),使得没有新的数据产生时，该代码会阻塞，直到有新的数据生成，再次进行处理
"""
pattern = '''(?P<remote>[d.]{7,}s-s-s[(?P<datetime>[^[]] )])s
"(?P<method>.*)s(?P<url>.*)s(?P<protocol>.*)"s(?P<status>d{3})s(?P<size>d )s"[^"] "s"(?P<useragent>[^"] )"'''
#编译
regex = re.compile(pattern)
#构造字典
ops = {
'datetime': lambda datestr: datetime.datetime.strptime(datestr, '%d/%b/%Y:%H:%M:%S %z'),
'status': int,
'size': int,
'useragent': lambda ua: parse(ua)
}
#提取信息
def extract(line: str) -> dict:
matcher = regex.match(line)
if matcher:
return {name: ops.get(name, lambda x: x)(data) for name, data in matcher.groupdict().items()}
# 打开文件
def openfile(path: str):
"""装载日志文件"""
with open(path) as f:
for line in f:
fields = extract(line)
if fields:
yield fields
else:
continue
#装载文件，判断文件类型已经是否存在
def load(*paths):
for item in paths:
p = Path(item)
if not p.exists():
continue
if p.is_dir():
for file in p.iterdir():
if file.is_file():
yield from openfile(str(file))
elif p.is_file():
yield from openfile(str(p))
# 随机生成100以内的数字
def source(second=1):
"""生成数据"""
while True:
yield {
'datetime': datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=8))),
'value': random.randint(1, 100)
}
time.sleep(second)
# 滑动窗口函数
def window(src: Queue, handler, width: int, interval: int):
'''
窗口函数，表示间隔一段时间取出一定的数据进行处理
:param src:数据源，这里是缓存队列，用于获取数据
：param handler:数据处理的函数
：param width:时间窗口函数，秒
：param interval:处理时间间隔，秒
'''
start = datetime.datetime.strptime('20170101 000000 0800', '%Y%m%d %H%M%S %z')
current = datetime.datetime.strptime('20170101 010000 0800', '%Y%m%d %H%M%S %z')
buffer = []
delta = datetime.timedelta(seconds=width - interval)
while True:
# 从数据源获取数据
data = src.get() # 这个代码会阻塞，等待数据输入，没有数据输入就阻塞
if data:
buffer.append(data)
current = data['datetime'] # 存入临时缓冲等待计算
# 每隔interval重新计算buffer中的一次数据
if (current - start).total_seconds() >= interval:
ret = handler(buffer)
start = current
# 清除超出width的数据
buffer = [x for x in buffer if x['datetime'] > current - delta]
# 随机数平均的测算函数
source()
def handler(iterable):
#return sum(map(lambda x: x['value'], iterable)) / len(iterable)
print(sum(map(lambda x:x['value'],iterable))/len(iterable))
# 测试函数
def donothing_handler(iterable):
#return iterable
print(iterable)
# 状态码占比
def status_handler(iterable):
# 时间窗口内的一批数据
status = {}
for item in iterable:
key = item['status']
status[key] = status.get(key, 0) 1
total = len(iterable)
print({k:float( "{:.2f}".format(status[k] / total)) for k, v in status.items()})
return {k: status[k] / total for k, v in status.items()}
# 浏览器分析
allbrowsers = {}
def browser_handler(iterable):
browsers = {}
for item in iterable:
ua = item['useragent']
key = (ua.browser.family, ua.browser.version_string)
browsers[key] = browsers.get(key, 0) 1
allbrowsers[key] = allbrowsers.get(key, 0) 1
print(sorted(allbrowsers.items(), key=lambda x: x[1], reverse=True)[:10])
return browsers
# 分发器
def dispatcher(src):
# 分发器中记录handler,同时保存各自的队列
handlers = []
queues = []
def reg(handler, width: int, interval: int):
"""
注册窗口处理函数
：param handler:注册数据处理函数
：param width:时间窗口宽度
：param interval:时间间隔
"""
q = Queue()
queues.append(q)
# 多线程，数据并行
h = threading.Thread(target=window, args=(q, handler, width, interval))
handlers.append(h)
def run():
# 启动线程处理数据
for t in handlers:
t.start()
# 将获取到的数据分发到所有的队列中
for item in src:
for q in queues:
q.put(item)
# print(q.get())
return reg, run
if __name__ == "__main__":
import sys
path = '/tmp/test.log'
"""
以下的代码为测试用的，用于统计每隔5s统计10s内的随机数字的平均值
reg, run = dispatcher(source())
reg(handler, 10, 5)
"""
reg, run = dispatcher(load(path))
#每隔5s返回过去10s的数据，但是不做处理
reg(donothing_handler, 10, 5)
#每隔5s统计10s内的返回状态码的占比情况
reg(status_handler, 10, 5)
# 每隔5s统计10s内的浏览器类型占比情况，展示排行10s内访问量前十的浏览器
reg(browser_handler,10,5)
run()
