【数据采集】基于flask框架的日志采集系统（附源代码）

我们在网站上的一切 *** 作在后台都会以日志的形式输出，日志就是行为记录。那我们怎么显式的看到我们的行为轨迹呢？就需要用到日志采集系统啦

本文会带你实现一个简易的日志采集系统

代码的运行效果如图：

目录

打开MySQL

建立数据库

LogDB.py 数据库 *** 作代码

demoLog.py 日志采集代码

dbProcess.py 定义数据表结构与 *** 作

logs.html 日志显示网页

index.html 登录首页 

flask搭建框架 

运行结果展示 

---

打开MySQL

采集之前，需要打开 MySQL，存储日志

这里有一个快速通道，不用下载 MySQL，我们直接下载打开小皮面板就行啦，当然也可以用别的

在代码运行过程中，要一直保持 MySQL 是开启状态

         在打开小皮面板的时候，会提示是否同意更改，点击 是

启动 MySQL 8.0.12，如果启动失败，可以尝试在配置中更改端口，概率启动成功

建立数据库

打开数据库功能。首先会要求我们更改 root 密码，更改之后，点击左上角创建数据库，创建一个新的数据库 test。取别的名字也可以，我们举例使用 test

数据库基本准备完成，我们来看看代码吧

LogDB.py 数据库 *** 作代码

这里 connect 的是我们建立的数据库，要填写数据库相关信息

此外，在库中，我们需要建立两个数据表 log123 与 dbuser，代码中只给出了 dbuser 表的定义方式，大家可根据需要自行定义 log123 表

log123 用于存储日志信息

dbuser 用于存储用户的登录账号与密码

import pymysql
from sklearn.utils import resample
class logDB0():
    def __init__(self):
        self.conn = pymysql.connect(host="127.0.0.1",
                                    port=3310,
                                    user="DB",
                                    password="123456",
                                    database="test"
                                    )
        self.cursor = self.conn.cursor()
    def ddLog(self,sql = None):
        _isSuccess = False
        try:
            self.cursor.execute(sql)
            self.conn.cursor()
            _isSuccess = True
        except:pass
        return _isSuccess
    def queryLog(self,sql = None):
        self.cursor.execute(sql)
        res = self.cursor.fetchall()
        return res
if __name__ =='__main__':
    d = logDB0()
    sqllsy = 'create table dbuser (id integer auto_increment primary key ,username varchar(255), userpwd varchar(255))'
    #sqllsy = 'select * from dbuser'
    print(sqllsy)
    print(d.queryLog(sql=sqllsy))

demoLog.py 日志采集代码

我们需要下载 logging 库，日志采集主要依靠 logging 库

import logging
import logging.config
import LogDB
from time import asctime
from datetime import datetime
db = LogDB.logDB0()

class MyLogHandler(logging.Handler, object):
    # 自定义日志 handler
    def __init__(self):
        logging.Handler.__init__(self)

    def emit(self, record):
        try:
            sql = "insert into log123 values(null,'{}','{}','{}')".format(str(datetime.now()), record.funcName, record.getMessage())
            db.ddLog(sql=sql)
        except Exception:
            self.handleError(record)

def mylog():
    # 创建一个日志记录器
    log = logging.getLogger('test_logger')
    log.setLevel(logging.DEBUG)
    # 创建一个日志处理器
    logHandler = MyLogHandler()
    logHandler.setLevel(logging.DEBUG)
    # 创建一个日志格式器
    formats = logging.Formatter('%(asctime)s-%(name)s-%(lineno)d:%(message)s')
    # 将日志格式器添加到日志处理器中
    logHandler.setFormatter(formats)
    # 将日志处理器添加到日志记录器中
    log.addHandler(logHandler)
    return log
logl = mylog()
logl.debug("hello,okok")

dbProcess.py 定义数据表结构与 *** 作

from sqlalchemy import create_engine, Column, Integer, String, MetaData
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
Base = declarative_base()

# 表结构定义
class Logs(Base):
    __tablename__ = 'logs';
    id = Column(Integer, primary_key=True, autoincrement=True)
    ctime = Column(String(100))
    clineno = Column(Integer)
    cfunname = Column(String(100))
    cmessage = Column(String(100))

#  *** 作类
class dbModel1():
    def __init__ (self):
        self.engine = create_engine('mysql+pymysql://用户名:数据库密码@localhost/数据库名')
        DBsession = sessionmaker(bind=self.engine)
        self.session = DBsession()

    # 连表
    def createTable(self):
        Base.meradata.create_all(self.engine)

    # 增加数据
    def insert(self, ct=None, linen=None, msg=None):
        logs = Logs(ctime=ct, clineno=linen, cmessage=msg)
        self.session.add(logs)
        self.session.commit()

    #查询数据
    def query(self):
        q = self.session.query(Logs)
        return q.all()

logs.html 日志显示网页

    
    




{% for item in data['res'] %}

{% endfor %}

    

    
序号
    
时间
    
程序名
    
具体信息
    

    
{{item[0]}}
    
{{item[1]}}
    
{{item[2]}}
    
{{item[3]}}

index.html 登录首页 

注：超链接到 logs.html 网页，用 flask 路由，不要用网页链接

        
        



Logging--Check

         
                 
         
         
                 
         
         
                
         
    

flask搭建框架 

from flask import Flask, render_template, request
from LogCollect import demoLog
from LogCollect import LogDB

app = Flask(__name__)
loger = demoLog.mylog()
db = LogDB.logDB0()

@app.route("/")
def index():
    loger.debug("请求网页")
    return  render_template('index.html')

@app.route('/login', methods=['POST'])
def login():
    username = request.form.get('username')
    userpwd = request.form.get('userpwd')
    dbinfo = db.queryLog('select * from dbuser where username="{}"'.format(username))
    if dbinfo:
        loger.debug("用户名正确")
        if userpwd == dbinfo[0][2]:
            print("hello! welcome")
            loger.debug("允许登录")
        else:
            print("用户名错误")
            loger.debug("不允许登录，请改正信息")
    else:
        print("用户名错误!")
        loger.debug("不允许登录，请改正信息!!")
@app.route('/logChecks')
def logList():
    data = {}
    data['res'] = db.queryLog("select * from log123")
    return render_template('logs.html', data=data)

if __name__ == '__main__':
    app.run()
    loger.debug("请求服务器.....")

运行结果展示