发表于 | 更新于: | 分类于 ， | | 阅读次数:
字数统计: 888 | 阅读时长 ≈ 3

建设关键信息基础设施应当确保其具有支持业务稳定、持续运行的性能，并保证安全技术措施同步规划、同步建设、同步使用。

–《网络安全法》第三十三条

信息系统的生命周期层面和安全保障要素层面不是相互孤立的，而是相互关联、密不可分的。
信息系统安全保障生命周期的安全保障要素
在信息系统生命周期模型中，将信息系统的整个生命周期抽象成计划组织、开发采购、实施交付、运行维护和废弃5个阶段，加上在运行维护阶段的变更产生的反馈，形成信息系统生命周期完整的闭环结构。在信息系统生命周期中的任何时间点上，都需要综合信息系统安全保障的技术、管理、工程和人员保障要素。从信息系统建设开始，就应该综合考虑系统的安全保障要求，确保信息系统建设和信息系统安全保障建设同步规划、同步实施。也就是我们平时讲的信息系统和安全保障要“三同步”–同步规划、同步建设、同步使用。

  • 同步规划

关键信息基础设施运营者应：
a）同步分析安全需求，即在关键信息基础设施建设或改建之初，从本组织的职能或业务的角度分析对关键信息基础设施实施网络安全的需求，形成安全需求说明书。
b）同步定义安全要求，即基于网络安全需求说明书，定义关键信息基础设施的网络安全要求，形成网络安全功能和性能说明书。
c）确保安全需求说明书得到网络安全责任部门签字认可。

  • 同步建设

关键信息基础设施运营者应：
a）同步设计安全体系结构，即基于已经定义的关键信息基础设施的网络安全要求，设计网络安全体系结构，明确系统内的各类信息安全组件，说明各组件提供的信息安全服务及可能的实现机制。
b）同步开展详细的安全设计，即根据安全保护等级选择基本安全措施，细化安全机制在关键信息基础设施中的具体实现。
c）在建设或改建过程中，按照GB/T 22239工程实施相应等级的要求，同步建设符合其等级要求的网络安全设施，包括自行软件开发。
d）建设完成后，组织对关键信息基础设施进行验收并将网络安全作为验收的重要内容。

  • 同步使用

关键信息基础设施运营者应：
a）同步运行安全设施，确保安全设施保持启用状态。
b）按照GB/T 22239安全运维管理相应等级的要求进行安全运维。
c）关键信息基础设施及其运行环境发生明显变化时，评估其风险，及时升级安全设施并实施变更管理。
d）对安全设施同步实施配置管理，包括制定配置管理计划，制定、记录、维护基线配置，保留基线配置的历史版本，便于必要时恢复历史配置。

博客地址：http://xiejava.ishareread.com/


“fullbug”微信公众号

关注：微信公众号,一起学习成长！

发表于 | 更新于: | 分类于 | | 阅读次数:
字数统计: 2.3k | 阅读时长 ≈ 9

最近有位做医疗项目的同学咨询有一批人员的身高、体重、性别、年龄、心电图、是否有心脏病等数据是否可以根据这些数据预测某个人是否有心脏病的迹象。这当然是可以的，AI机器学习不就是干这事的吗？这是一个典型的分类算法。根据这些人体特征来判断是否存在潜在的疾病。问题是如何对心电图进行特征提取，提取出相关的特征，让模型进行学习和训练。

拿到数据后，第一步，我们首先来看一下数据。

一、数据探索

1、读入数据

导入numpy、pandas、matplotlib三大件，读入数据

1
2
3
4
5
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
df=pd.read_csv('data_csv.csv')
df

在这里插入图片描述

数据集中有age年龄、ismale性别、Weight身高、Height体重、HRTrend心电图、label是否有心脏病（1，是有心脏病、0是无心脏病）
这里看到主要的几个关键性的指标。HRTrend是csv中载入的时候数据是文本字符串格式，还有nan及\n等字符串，我们要将这些数据进行清洗规则化，将其转化为数组。

2、数据处理

先拿一个样本数据来看看数据的情况。对它进行相应的处理，包括去掉回车换行符、去多余空格、然后以空格作为分隔形成数组。

1
2
3
4
5
6
7
HRTrend_Data=df['HRTrend'][1][1:len(df['HRTrend'][1])-1]
HRTrend_Data=HRTrend_Data.replace('\n','') #替换掉换行符
HRTrend_Data=HRTrend_Data.replace('nan',str(0))   #将nan补0
HRTrend_Data=re.sub(' +', ' ', HRTrend_Data).strip() #去掉多余的空格
HRTrend_Data=np.asarray([float(s) for s in HRTrend_Data.split(' ')])  #将字符串通过空格分隔，转换成数组
pd.DataFrame(HRTrend_Data).plot()
HRTrend_Data

在这里插入图片描述

这个图形不是很规则不平滑，看数据是因为异常值太多，后面的数据都是0，这些0都是因为数据为Non补充进来的，有可能是应为数据采集的时候某个点位没有采集到。我们需要将这些异常值给处理掉。
对于中间为0的，我们可以通过采用前值填充或者后值填充（也就是用前面的值或后面的值来替代为0的值），后面全部为0的部分要去掉。应为会影响到一些关键性的特征，如均值、50%的值、70%的值等。

1
2
df_HRTrend1=pd.DataFrame(HRTrend_Data)
df_HRTrend1.describe()

在这里插入图片描述

可以看到，这些异常值将会很大程度的影响一些关键特征，所以我们要处理异常值。

1
2
3
4
5
HRTrend_Data=np.trim_zeros(HRTrend_Data)  #去掉首尾的0
df_data1=pd.DataFrame(HRTrend_Data)
df_data1.replace(to_replace=0, method='ffill',inplace=True)  #用前值替换掉中间的异常0
df_data1.plot()
df_data1.describe()

在这里插入图片描述

把刚对一个样本处理的过程写成一个函数，应用到所有的样本数据。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def procdata(df_dataclumn):
    ary_data=df_dataclumn[1:len(df_dataclumn)-1]
    ary_data=ary_data.replace('\n','')
    ary_data=ary_data.replace('nan',str(0))
    ary_data=re.sub(' +', ' ', ary_data).strip()
    ary_data=np.asarray([float(s) for s in ary_data.split(' ')])
    ary_data=np.trim_zeros(ary_data)
    df_data=pd.DataFrame(ary_data)
    df_data.replace(to_replace=0, method='ffill',inplace=True)
    ary_data=df_data.values
    return ary_data

df['HRTrend']=df['HRTrend'].map(procdata)
df['Weight'].replace(to_replace=0, method='ffill',inplace=True) #去异常值
df['Height'].replace(to_replace=0, method='ffill',inplace=True) #去异常值
df

在这里插入图片描述
找前六个样本看看效果

1
2
3
4
5
6
fig,axes = plt.subplots(2,3,figsize=(12,4))
i=0
for x in range(2):
    for y in range(3):
        i=i+1
        axes[x,y].plot(df['HRTrend'][i])

在这里插入图片描述

二、特征工程及模型训练

方法一：通过描叙统计提取时序特征并进行模型训练

  • 特征提取

有量纲时序特征提取包括以内容。
在这里插入图片描述

可以看到，大部分内容都可以通过describe()获取。可以写个函数方法来获取相关的特征值。

1
2
3
4
5
6
7
8
9
10
def getdescribefeatures(data):
    df_feature=pd.DataFrame(data)
    df_feature=df_feature.describe().T
    range=df_feature['max'][0]-df_feature['min'][0]
    return range,df_feature['count'][0],df_feature['mean'][0],df_feature['std'][0],df_feature['min'][0],df_feature['25%'][0],df_feature['50%'][0],df_feature['75%'][0],df_feature['max'][0]

HRTrend_np=np.array(df['HRTrend'])
HRTrend_features_ary=[getdescribefeatures(d) for d in HRTrend_np]
df_HRTrend_features=pd.DataFrame(HRTrend_features_ary,columns=['HRTrend'+str(i) for i in range(len(HRTrend_features_ary[0]))])
df_HRTrend_features

在这里插入图片描述

将心电图提出的统计信息特征与年龄、性别、身高、体重等特征合并，形成特征集。

1
df_data=df[['age','ismale','Weight','Height','label']].join(df_HRTrend_features)

预留15个样本做最后的预测看效果。

1
2
df_train_data=df_data[0:len(df_data)-15]    #训练数据
df_predict_data=df_data[len(df_data)-15:len(df_data)]  #用于预测的数据
  • 模型训练
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from sklearn.metrics import f1_score
from sklearn.metrics import recall_score
from sklearn.metrics import accuracy_score
from sklearn.svm import SVC
from sklearn.neighbors import KNeighborsClassifier

x_tt, x_validation, y_tt, y_validation = train_test_split(x, y, test_size=0.2)
# 将训练集再切分为训练集和测试集
x_train, x_test, y_train, y_test = train_test_split(x_tt, y_tt, test_size=0.25)

#将模型的名字和模型函数作为元组添加到列表当中存储；
models = []
models.append(("KNN",KNeighborsClassifier(n_neighbors=3))) #指定邻居个数
models.append(("SVM Classifier",SVC(C=1000)))
#可以通过参数C来控制精度，C越大要求精度越高；  C——错分点的惩罚度

#循环调用所有模型进行训练、预测
for clf_name, clf in models:
    clf.fit(x_train, y_train)
    xy_lst = [(x_train, y_train), (x_validation, y_validation), (x_test, y_test)]
    for i in range(len(xy_lst)):
        x_part = xy_lst[i][0]  # 为遍历中的第0部分
        y_part = xy_lst[i][1]  # 为遍历中的第1部分
        y_pred = clf.predict(x_part)
        print(i)  # i是下标，0表示训练集，1表示验证集，2表示测试集
        print(clf_name, "ACC:", accuracy_score(y_part, y_pred))
        print(clf_name, "REC:", recall_score(y_part, y_pred))
        print(clf_name, "F-score:", f1_score(y_part, y_pred))

结果如下：
在这里插入图片描述

SVM分类模型的效果要稍好于KNN的模型，但整体效果还不是很好。应该是提取的特征还是不够。我们来试下通过tsfresh提取时序特征并进行模型训练。

方法二：通过tsfresh提取时序特征并进行模型训练

  • 特征提取

tsfresh可以自动提取各种时间特征，可以参考官方文档，一般的取最大、最小、偏度、峰度等统计指标这些都自动化集成了。
https://tsfresh.readthedocs.io/en/latest/text/introduction.html
在这里插入图片描述

通过tsfresh提取时序特征代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
from tsfresh import extract_features
from tsfresh import select_features
dfx_HRTrend=pd.DataFrame()
for i in df.index:
    df_i=pd.DataFrame(df['HRTrend'][i],columns=['HRTrend'])
    df_i['id']=i
    df_i['time']=np.array(range(1,len(df_i)+1))
    df_i['label']=df['label'][i]
    dfx_HRTrend=pd.concat(objs=[dfx_HRTrend,df_i])

dfx_xtracted_features = extract_features(dfx_HRTrend,column_id="id", column_sort="time")
dfx_xtracted_features

在这里插入图片描述

可以看到tsfresh自动提取了1574colums的特征，但是我们要选取与标签有相关性的特征。tsfresh可以自动做到这一点，通过tsfresh的select_features传入特征和标签来选择与标签有相关性的特征，

1
2
3
4
from tsfresh.utilities.dataframe_functions import impute
impute(dfx_xtracted_features)
df_HRTrend_features = select_features(dfx_xtracted_features, df['label'])
df_HRTrend_features

在这里插入图片描述

通过特征选择自动选取了有相关性的143个特征，比前面缩减了很多，有利于提高模型训练的效率和精度。
我们再来看一下模型训练的效果。
同样将心电图通过tsfresh提出的时序特征与年龄、性别、身高、体重等特征合并，形成特征集。

1
df_data=df[['age','ismale','Weight','Height','label']].join(df_HRTrend_features)

预留15个样本做最后的预测看效果。

1
2
df_train_data=df_data[0:len(df_data)-15]    #训练数据
df_predict_data=df_data[len(df_data)-15:len(df_data)]  #用于预测的数据

模型训练

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from sklearn.metrics import f1_score
from sklearn.metrics import recall_score
from sklearn.metrics import accuracy_score
from sklearn.svm import SVC
from sklearn.neighbors import KNeighborsClassifier

x_tt, x_validation, y_tt, y_validation = train_test_split(x, y, test_size=0.2)
# 将训练集再切分为训练集和测试集
x_train, x_test, y_train, y_test = train_test_split(x_tt, y_tt, test_size=0.25)

#将模型的名字和模型函数作为元组添加到列表当中存储；
models = []
models.append(("KNN",KNeighborsClassifier(n_neighbors=3))) #指定邻居个数
models.append(("SVM Classifier",SVC(C=1000)))
#可以通过参数C来控制精度，C越大要求精度越高；  C——错分点的惩罚度

#循环调用所有模型进行训练、预测
for clf_name, clf in models:
    clf.fit(x_train, y_train)
    xy_lst = [(x_train, y_train), (x_validation, y_validation), (x_test, y_test)]
    for i in range(len(xy_lst)):
        x_part = xy_lst[i][0]  # 为遍历中的第0部分
        y_part = xy_lst[i][1]  # 为遍历中的第1部分
        y_pred = clf.predict(x_part)
        print(i)  # i是下标，0表示训练集，1表示验证集，2表示测试集
        print(clf_name, "ACC:", accuracy_score(y_part, y_pred))
        print(clf_name, "REC:", recall_score(y_part, y_pred))
        print(clf_name, "F-score:", f1_score(y_part, y_pred))

结果如下：
在这里插入图片描述

可以看到不管是KNN模型和SVM的模型都取得了不错的效果。

三、效果

最后我们来看一下，用预留的15个样本预测的效果。

1
2
3
4
5
6
7
8
9
10
11
pre_test_y=df_predict_data['label']
pre_test_x=df_predict_data.drop(columns='label')

for clf_name, clf in models:
    for i in pre_test_x.index:
        y_predict=clf.predict(pre_test_x.loc[[i]])
        predict_result='预测错误！'
        if y_predict==pre_test_y.loc[[i]].values:
            predict_result='预测正确！'
        print('第'+str(i)+'个样本'+clf_name+' 预测y='+str(y_predict)+'--实际 y='+str(pre_test_y.loc[[i]].values)+' - '+predict_result)
    print('-'*10)

先是通过描叙统计提取时序特征并进行模型训练后模型预测的结果：
在这里插入图片描述

还是有不少预测错误了。
再来看通过tsfresh提取时序特征并进行模型训练后模型预测的结果：
在这里插入图片描述

可以看到预留的15个样本都预测正确，取得了不错的效果！

本文通过心电图分类模型介绍了从数据处理、时序特征提取、模型训练和效果评估。验证了对于小数据量，通过tsfresh提取时序特征，往往能够取得很好的效果。

全部源代码及数据集见：https://download.csdn.net/download/fullbug/87369380

博客地址：http://xiejava.ishareread.com/


“fullbug”微信公众号

关注：微信公众号,一起学习成长！

发表于 | 更新于: | 分类于 ， | | 阅读次数:
字数统计: 2.7k | 阅读时长 ≈ 11

用Django框架进行web开发非常的快捷方便，但Django框架请求/响应是同步的。但我们在实际项目中经常会碰到一些耗时的不能立即返回请求结果任务如：数据爬取、发邮件等，如果常时间等待对用户体验不是很好，在这种情况下就需要实现异步实现，马上返回响应请求，但真正的耗时任务在后台异步执行。Django框架本身无法实现异步响应但可以通过Celery很快的实现异步和定时任务。本文将介绍如何通过Django+Celery+Flower实现异步和定时任务及其任务的监控告警。

常见的任务有两类，一类是异步任务，一类是定时任务（定时执行或按一定周期执行）。Celery都能很好的支持。

Celery 是一个 基于python开发的分布式异步消息任务队列，通过它可以轻松的实现任务的异步处理， 如果你的业务场景中需要用到异步任务，就可以考虑使用celery， 举几个实例场景中可用的例子:

  • 异步任务：将耗时的操作任务提交给Celery去异步执行，比如发送短信/邮件、消息推送、音频处理等等
  • 做一个定时任务，比如每天定时执行爬虫爬取指定内容

Celery 在执行任务时需要通过一个消息中间件（Broker）来接收和发送任务消息，以及存储任务结果， 一般使用rabbitMQ、Redis或其他DB。

本文使用redis作为消息中间件和结果存储，在后面的通过数据库监控任务执行案例将介绍用到数据库作为结果存储。

一、在Django中引入Celary

1、安装库

1
2
3
pip install celery
pip install redis
pip install eventlet  #在windows环境下需要安装eventlet包

2、引入celary

在主项目目录下，新建celary.py文件，内容如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
import os
import django
from celery import Celery
from django.conf import settings

# 设置系统环境变量，否则在启动celery时会报错
# taskproject 是当前项目名
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'taskproject.settings')
django.setup()

celery_app = Celery('taskproject')
celery_app.config_from_object('django.conf:settings')
celery_app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

在这里插入图片描述

在主目录的init.py中添加如下代码：

1
2
3
from .celery import celery_app

__all__ = ['celery_app']

在这里插入图片描述

3、在settings.py中设置celery的相关参数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
###----Celery redis 配置-----###
# Broker配置，使用Redis作为消息中间件
BROKER_URL = 'redis://:redispassword@127.0.0.1:6379/0'

# BACKEND配置，使用redis
CELERY_RESULT_BACKEND = 'redis://:redispassword@127.0.0.1:6379/1'


CELERY_ACCEPT_CONTENT=['json']
CELERY_TASK_SERIALIZER='json'
# 结果序列化方案
CELERY_RESULT_SERIALIZER = 'json'

# 任务结果过期时间，秒
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24

# 时区配置
CELERY_TIMEZONE = 'Asia/Shanghai'

在这里插入图片描述

这时候Celery的基本配置完成了，可以实现并添加任务了。

二、实现异步任务

1、创建tasks.py

在子应用下建立各自对应的任务文件tasks.py(必须是tasks.py这个名字，不允许修改)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import datetime
from time import sleep
from celery import shared_task
import logging
logger = logging.getLogger(__name__)


@shared_task()
def task1(x):
    for i in range(int(x)):
        sleep(1)
        logger.info('this is task1 '+str(i))
    return x


@shared_task
def scheduletask1():
    now = datetime.datetime.now()
    logger.info('this is scheduletask '+now.strftime("%Y-%m-%d %H:%M:%S"))
    return None

在tasks.py中我们定义了两个任务，这两个任务要用@shared_task装饰起来，否则celery无法管理。
在这里插入图片描述

为了放便执行我们通过views把这两个任务通过函数方法调用起来，用URL进行发布。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
views.py
from django.http import JsonResponse
from . import tasks
# Create your views here.


def runtask(request):
    x=request.GET.get('x')
    tasks.task1.delay(x)
    content= {'200': 'run task1 success!---'+str(x)}
    return JsonResponse(content)


def runscheduletask(request):
    tasks.scheduletask1.delay()
    content= {'200': 'success！'}
    return JsonResponse(content)

在这里插入图片描述

在urls中加入路由进行发布

1
2
3
4
5
6
7
8
from django.urls import path

from taskapp import views

urlpatterns = [
    path('task', views.runtask),
    path('runscheduletask', views.runscheduletask),
]

在这里插入图片描述

在项目的主urls中加入子项目的urls
在这里插入图片描述

2、启动celery

在启动celery之前，先要启动redis服务，因为celery在settings中配置要用到redis作为消息中间件和结果存储。
windows环境下启动redis的命令为redis-server.exe redis.windows.conf

在控制台启动celery的worker

1
celery -A taskproject worker -l debug -P eventlet

在这里插入图片描述

启动django访问url调用任务，看异步效果
在这里插入图片描述

3、查看任务

控制台查看异步任务执行的情况，可以看web的url很快返回响应结果，后台控制台一直在执行异步任务。
在这里插入图片描述

三、实现定时任务

Celery实现定时任务也很方便

1、定义调度器

在settings.py中加入定时任务的定义就可以实现定时任务

1
2
3
4
5
6
7
8
9
10
11
from celery.schedules import crontab

CELERYBEAT_SCHEDULE = {
    'every_5_seconds': {
        # 任务路径
        'task': 'taskapp.tasks.scheduletask1',
        # 每5秒执行一次
        'schedule': 5,
        'args': ()
    }
}

这里这个scheduletask1是前面tasks.py中定义的任务，当然也可以定义多个定时任务，如加一个task1，task1是有参数的，可以在’args’: ()中传入参数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
CELERYBEAT_SCHEDULE = {
    'every_5_seconds': {
        # 任务路径
        'task': 'taskapp.tasks.scheduletask1',
        # 每5秒执行一次
        'schedule': 5,
        'args': ()
    },
    'every_10_seconds': {
        # 任务路径
        'task': 'taskapp.tasks.task1',
        # 每10秒执行一次,task1的参数是5
        'schedule': 10,
        'args': ([5])
    }
}

在这里插入图片描述

这里定义了task1是10秒执行一次，传入的参数是5。

2、启动beat

需要保持worker进程，另外开一个控制台启动beat

1
celery -A taskproject beat -l debug

3、查看任务

启动任务后看控制台打印的日志task1和scheduletask1都按计划定时执行了。
在这里插入图片描述

三、通过数据库配置定时任务

虽然通过settings.py的配置可以实现定时任务的配置，做为实际项目中可能还是不够实用，更加工程化的做法是将定时任务的配置放到数据库里通过界面来配置。同样Celery对此也提供了很好的支持，这需要安装django-celery-beat插件。以下将介绍使用过程。

1、安装djiango-celery-beat

1
pip install django-celery-beat

2、在APP中注册djiango-celery-beat

1
2
3
4
INSTALLED_APPS = [
....
'django_celery_beat',
]

3、在settings.py中设置调度器及时区

在settings.py中屏蔽到原来的调度器，加入

1
CELERYBEAT_SCHEDULER = 'django_celery_beat.schedulers.DatabaseScheduler'

在这里插入图片描述

在setings.py中设置好语言、时区等

1
2
3
4
5
6
7
LANGUAGE_CODE = 'zh-hans'

TIME_ZONE = 'Asia/Shanghai'

USE_I18N = True

USE_TZ = False

4、进行数据库迁移

1
python manage.py migrate django_celery_beat

5、分别启动woker和beta

在两个控制台分别启动woker和beta

1
celery -A taskproject worker -l debug -P eventlet
1
celery -A taskproject beat -l debug

6、启动django服务，访问admin的web管理端

访问 http://localhost:8000/admin/ 可以看到周期任务的管理菜单，管理定时任务非常方便。
在这里插入图片描述

7、配置定时任务

点击“间隔”
在这里插入图片描述

点击“增加间隔”来增加定时任务的配置，增加一个5秒执行一次的定时器。
在这里插入图片描述

看到有个每5秒的定时器
在这里插入图片描述

这时可以用这个定时器去新建调度任务了。选择周期性任务，点击“增加周期性任务”
在这里插入图片描述

填入任务名，选择需要定时执行的任务
在这里插入图片描述

因为task1需要参数，在后面参数设置中进行参数的设置。
在这里插入图片描述

保存后可以看到新加了一条“每5秒执行一次task1”的调度任务。
在这里插入图片描述

8、查看调度效果

在woker和beta的控制台都可以看到有定时任务执行的信息，说明任务被成功调度执行了。
在这里插入图片描述

四、通过django的web界面监控任务执行情况

在控制台监控任务执行情况，还不是很方便，最好是能够通过web界面看到任务的执行情况，如有多少任务在执行，有多少任务执行失败了等。这个Celery也是可以做到了，就是将任务执行结果写到数据库中，通过web界面显示出来。这里要用到django-celery-results插件。通过插件可以使用Django的orm作为结果存储，这样的好处在于我们可以直接通过django的数据查看到任务状态，同时为可以制定更多的操作，下面介绍如何使用orm作为结果存储。

1、安装django-celery-results

1
pip install django-celery-results

2、配置settings.py，注册app

1
2
3
4
INSTALLED_APPS = (
...,
'django_celery_results',
)

3、修改backend配置，将Redis改为django-db

1
2
3
4
# BACKEND配置，使用redis
#CELERY_RESULT_BACKEND = 'redis://:12345678@127.0.0.1:6379/1'
# 使用使用django orm 作为结果存储
CELERY_RESULT_BACKEND = 'django-db'  #使用django orm 作为结果存储

4、迁移数据库

1
python manage.py migrate django_celery_results

可以看到创建了django_celery_results相关的表
在这里插入图片描述

5、查看任务

启动django服务后，执行异步和定时任务，就可以在管理界面看到任务的执行情况，执行了哪些任务，哪些任务执行失败了等。
在这里插入图片描述

五、通过Flower监控任务执行情况

如果不想通django的管理界面监控任务的执行，还可以通过Flower插件来进行任务的监控。FLower的界面更加丰富，可以监控的信息更全。以下介绍通过Flower来进行任务监控。

1、安装flower

1
pip install flower

2、启动flower

1
celery -A taskproject flower --port-5566

在这里插入图片描述

3、使用flower进行任务监控

在这里插入图片描述

点击失败的我们可以看到执行失败的详情，这里是故意给task1的参数传了个‘a’字符，导致它执行报错了。可以看到任务执行的报错信息也展示出来了。
在这里插入图片描述

六、实现任务异常自动邮件告警

虽然可以通过界面来监控了，但是我们想要得更多，人不可能天天盯着界面看吧，如果能实现任务执行失败就自动发邮件告警就好了。这个Celery当然也是没有问题的。
通过钩子程序在异常的时候触发邮件通知。

1、加入钩子程序

对tasks.py的改造如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
import datetime
from time import sleep
from celery import shared_task
from celery import Task
from django.core.mail import send_mail
import logging
logger = logging.getLogger(__name__)

class MyHookTask(Task):
    def on_success(self, retval, task_id, args, kwargs):
        info=f'任务成功-- 0task id:{task_id} , arg:{args} , successful !'
        logger.info(info)
        send_mail('celery任务监控', info, 'sendmail@qq.com', ['tomail@qq.com'])

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        info=f'任务失败-- task id:{task_id} , arg:{args} , failed ! erros: {exc}'
        logger.info(info)
        send_mail('celery任务监控异常', info, 'sendmail@qq.com', ['tomail@qq.com'])

    def on_retry(self, exc, task_id, args, kwargs, einfo):
        logger.info(f'task id:{task_id} , arg:{args} , retry !  erros: {exc}')


@shared_task(base=MyHookTask, bind=True)
def task1(self,x):
    for i in range(int(x)):
        sleep(1)
        logger.info('this is task1 '+str(i))
    return x


@shared_task
def scheduletask1():
    now = datetime.datetime.now()
    logger.info('this is scheduletask '+now.strftime("%Y-%m-%d %H:%M:%S"))
    return None

在这里插入图片描述

2、重启服务

将work和beta服务关掉，在两个控制台分别重新启动woker和beta

1
celery -A taskproject worker -l debug -P eventlet
1
celery -A taskproject beat -l debug

3、验证效果

在任务成功或失败的时候发邮件通知。
在这里插入图片描述

任务执行成功通知
在这里插入图片描述

任务执行异常告警通知
在这里插入图片描述

Django如何发送邮件见 https://blog.csdn.net/fullbug/article/details/128495415

至此，本文通过几个简单的应用介绍了Django+Celery+Flower实现异步和定时任务及其监控告警。

博客地址：http://xiejava.ishareread.com/


“fullbug”微信公众号

关注：微信公众号,一起学习成长！

发表于 | 更新于: | 分类于 | | 阅读次数:
字数统计: 926 | 阅读时长 ≈ 3

通过Django发送邮件非常的简单，在Python中已经内置了一个smtplib邮件发送模块，Django在此基础上进行了简单地封装，我们可以在Django的环境中方便的发送邮件。大部分邮件发送成功主要是邮件的参数配置，本文以常用的126邮箱和QQ邮箱为例介绍Django发送邮件的配置，其他提供smtp邮件服务的邮箱都是一样的。

一、Django邮件配置

用Django发送邮件需要告诉你的邮件服务器相应的参数配置，需要在settings.py中进行配置好。
默认情况下，使用配置文件中的EMAIL_HOST和EMAIL_PORT设置SMTP服务器主机和端口，EMAIL_HOST_USER和
EMAIL_HOST_PASSWORD是用户名和密码。如果设置了EMAIL_USE_TLS和EMAIL_USE_SSL，它们将控制是否使用相应的加密链接。
典型的配置如下：

1
2
3
4
5
EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_HOST = 'smtp.126.com'      #126邮箱的邮箱服务地址
EMAIL_PORT = 25          #端口，大部分都是默认25
EMAIL_HOST_USER = 'yourmail@126.com'    #这里是你的邮箱账号
EMAIL_HOST_PASSWORD = 'yourpassword'   #注意这里不能用你邮箱账号的密码，而要用申请的设备授权码。

这里要注意的两个地方，一个是EMAIL_PORT端口、一个是EMAIL_HOST_PASSWORD密码。
端口一般默认是25，但有些邮箱改了默认端口或需要用加密链接465、578端口，可以从邮件服务商查到。
密码以前可以用邮箱账号密码，但这几年随着安全要求的提高，大部分主要的邮件服务提供商都要用申请的授权码进一步加强安全系数。
这里介绍这些关键的信息如何在邮件服务商获取。

注意EMAIL_HOST_PASSWORD这里不能用你邮箱账号的密码，而要用申请的设备授权码。

126邮箱

进入到邮箱配置界面，点击“POP3/SMTP/IMAP”，注意要开启SMTP。
126邮箱配置界面

授权码是用于登录第三方邮件客户端的专用密码，点击”新增授权密码”，降会弹出一个账号安全验证，扫描后可以快速的发短信。
账号安全验证界面

用你的注册手机发完短信后点击”我已发送”就会弹出一个授权码的窗口，要把这个授权码记下来，这个窗口只显示一次。
授权码显示界面

QQ邮箱

进入到QQ邮箱WEB界面后点击“设置”，点击“邮件设置”的“账户”页签拉到“账户安全”处就可以看到相关的SMTP服务信息

QQ邮箱配置界面

点击“生成授权码”，将弹出验证密保的界面。
验证密保界面
根据提示发送短信后，点击我已发送，将会弹出生成授权码的界面显示授权码。
授权码生成界面

如果需要SSL加密方式，可以进一步查看服务商的SMTP服务说明
如QQ邮箱的服务说明 https://service.mail.qq.com/cgi-bin/help?subtype=1&&id=28&&no=331
服务说明界面

看到使用SSL，端口号是465或587

1
2
3
4
5
6
EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_HOST = 'smtp.qq.com'  #QQ邮箱的smtp服务器
EMAIL_PORT = 465           #端口为465或587
EMAIL_USE_SSL = True    #SSL加密方式设置为True
EMAIL_HOST_USER = 'yourmail@qq.com'     #这里是你的邮箱账号
EMAIL_HOST_PASSWORD = 'yourpassword'   #注意这里不能用你邮箱账号的密码，而要用申请的设备授权码。

二、测试发送邮件

可以用django的环境在shell中测试。

1
2
3
python manage.py shell
from django.core.mail import send_mail
send_mail('test','this is test mail','yourmail@qq.com',['yourmail@126.com','yormail@qq.com'])

在这里插入图片描述

如果反回1就表示发送成功了, 查收邮件到达。

如果发送不成功，请仔细核对邮箱服务的相关配置，一般来说与django无关。

博客地址：http://xiejava.ishareread.com/


“fullbug”微信公众号

关注：微信公众号,一起学习成长！

发表于 | 更新于: | 分类于 ， | | 阅读次数:
字数统计: 1k | 阅读时长 ≈ 3

近年来，高新技术数字经济快速发展，数字化转型成为各个行业高质量发展的重要引擎和创新路径。网络安全作为数字化建设的安全基石，却面临着网络安全人才缺口不断攀升的直接挑战。各安全公司依托自己网络安全经验的人才优势向客户提供各类安全服务，最常见的就是安全托管和重保安全。

安全托管服务与重保安全服务都是为客户提供安全服务，这两者有啥区别呢？

一、安全托管服务

安全托管服务（Managed Security Service，MSS），是由Gartner于2011年提出，简单来讲，即网络安全厂商通过自身的安全运营服务，包括人员、工具、平台等，为其用户提供7*24小时的安全托管服务。MSS安全托管服务最直接的目的，就是解决用户自身安全能力不足的问题，将专业的事情交给专业的人来做。说白了其实就是企业将自身的安全运营外包给有能力的安全厂商，解决自身安全运营能力不足的问题。

二、重保安全服务

重保安全服务（Cybersecurity in Important Period, CIP），是指在特殊时期（如HVV等）、重要活动（如两会等）、重大节日（如国庆等）期间为用户构建全方面的重要敏感时期的安全保障服务。保障网络基础设施、重点网站和业务系统安全，提供全方位的安全防守建设咨询以及事前、事中、事后的全面安全建设托管服务，确保企业客户的业务系统能够在重大活动期间安全平稳运行。

三、安全托管服务与重保安全服务的区别

从安全托管服务与重保安全服务的概念和定义上，个人理解安全托管服务与重保安全服务还是有所区别的。

一般来说，安全托管服务(MSS)服务侧重于管理和运营，以保障企业IT业务稳定运行为目的，主要是日常的安全运营。而重保安全更接近实战化主要侧重于重保期间的防护值守，有重点的企业客户的业务系统能够在重大活动期间安全平稳运行。

拿平时我们见到社会安保来类比，日常安全运营就是我们平时看到的日常治安巡逻；重保就是比如国庆期间、两会其间有针对性的加大安保防范力度如特警武装上岗执勤、某些重要场所限制人流、加强舆情监控等必要的防控手段。

  • 从具体的服务内容来看：

安全托管服务主要是日常安全运营相关的内容如：资产的定期梳理、暴露面的检测、定期的漏扫、漏扫后的安全加固整改、安全监测、事件协助处置、安全策略定期优化等。

重保安全服务主要是针对重保时期有重点的安全保障更贴近实战化如：准备阶段的资产清查、暴露面识别、暴露面收敛等；重保初期的重点系统（或靶标系统）专项加固、情报收集、攻击阻断演练等；重保阶段的安全事件分析、响应处置、溯源反制、防守报告等。

  • 从人员配置来看：

重保安全服务的人员配置不管是从人员数量还是个人网络安全专业水平来看都要比安全托管服务的高。一般来说重保安全服务的核心人员都会要求有攻防对抗能力。可以想像一下普通安保人员与武装特警的区别。

博客地址：http://xiejava.ishareread.com/


“fullbug”微信公众号

关注：微信公众号,一起学习成长！

发表于 | 更新于: | 分类于 ， | | 阅读次数:
字数统计: 2.6k | 阅读时长 ≈ 8

近年安全圈XDR的概念非常火，是近几年热点的安全技术之一，究竟什么是XDR，XDR核心能力是什么？能够解决什么问题，XDR与EDR、NDR关系和区别又是什么？让我们带着这些问题一起来认识一下XDR。

一、什么是XDR

XDR 于 2018 年由 Palo Alto 首席技术官 Nir Zuk 提出，2020-2021 年连续入选 Gartner 端点安全、安全运营技术成熟度曲线，目前处于创新启动期。

XDR全名是Extended Detection and Response（扩展检测和响应），因为缩写与EDR重名了，所以就取了Extended第二个字母X，缩成了XDR。

Gartner给出的XDR定义为：XDR是一种基于SaaS的、绑定于特定供应商的安全威胁检测和事件响应工具，它将多个安全产品集成到一个统一了所有许可安全组件的内聚安全运营系统中。

通常情况下，可以认为XDR是一个融合了多种安全检测、响应能力的平台框架，只要是为了解决威胁检测与响应的问题能力模块，都可以往里装。它需要将多个安全产品能力有机的结合在一起，有统一的数据格式、策略、交互界面。相对于EDR（端点的检测与响应）和NDR（网络的检测与响应）来说，特别的强调X的概念，也就是可以扩展（Extended）的检测与响应。这里X覆盖了云、网、端、威胁情报等，EDR和NDR及其他的检测设备都可以作为XDR的能力模块为XDR提供数据来源和检测手段。

二、XDR解决什么问题

近期， Gartner正式发布了2022安全运营技术成熟度曲线（Hype Cycle），正如大家所预测的那样，XDR终于站上了Peak of Inflated Expectations的顶端，成为安全运营体系中最炙手可热的技术之一，具体如下图所示：
2022安全运营技术成熟度曲线

那么在企业安全运营过程中究竟有什么困惑，XDR又能解决什么问题呢？

1、安全运营的困惑

  • 单兵作战/数据孤岛

企业虽然有了SIEM/SOC等日志类数据分析平台，或是IDS、IPS、WAF、防火墙、EDR等单点安全设备，但前者无法理解下游检测设备告警，数据多而不准，安全误报多；后者获取的数据又有限，不同设备数据还无法紧密集成，最后变成了真实风险看不到，出现威胁防不了。

  • 告警疲劳，误报多

安全人员每天都会收到来自不同安全设备的上万条威胁告警，而头疼的是，绝大多数（90%以上）都并非真实威胁，所以安全人员不是身体在处理误报的路上，就是精神在遭受误报的折磨，压力非常之大。

  • 全局态势不可见

企业虽然买了一堆的安全设备，这些设备都产生了相应的告警，但是没有形成统一的全局的安全风险态势，对于安全主管来说要重点防范哪些地方，企业的安全建设还有哪些短板，这些都不可以知不可见。

  • 响应处置能力弱

发生网络安全事件以后，无法有效溯源，快速的处置，尤其是多设备的联动处置，大多还停留在手工处置的阶段，处置效率先对比较低，不及时。

2、XDR的作用

XDR通过统一的交互框架、统一的数据标准、统一的数据存储方式进行安全数据采集、安全威胁集中分析、安全事件统一处置、响应编排。

XDR的核心作用在于能够跨越不同数据源与IT架构，集中汇集云、网、端、威胁情报等多源安全数据/工具。通过大数据与人工智能、用户行为分析等智能分析手段，对安全数据/事件进行关联分析，还原攻击路径，达到对整个攻击面的全面可视，解决安全孤岛的问题。基于动态更新的事件库与预置处置场景将产出的告警进行自动化编排与分诊，实现自动化响应。解决安全运营过程中数据孤岛、告警疲劳、全局态势不可见、响应处置能力弱的问题，更重要的是将企业安全运营水平和标准化产品挂钩，而非依赖不稳定的个人技术水平。

XDR的作用

三、XDR与EDR、NDR的区别

EDR从端点侧做威胁检测，确实能检测到攻击的准确信息，但是端点检测这种方式需要在用户主机上安装检测agent程序，无法覆盖用户所有的资产。并且端点检测的部署成本相比网络检测也更高，对于端点的操作系统、硬件配置、网络情况都有要求。

NDR从网络侧做威胁检测，检测到的更多都是攻击的特征或者攻击意图，此时攻击很有可能并未真正发生，或者并未造成严重后果。如果全部转化为威胁事件，则会造成告警风暴，给运营带来困难。

所以EDR的特点是检测的深但是覆盖面窄，而NDR的特点是检测的浅但是覆盖面广。

XDR则结合了这两者的全部优点，对于重点资产可采用端点检测方式，对于其他资产可采用网络检测方式。XDR平台会将这两种能力检测到的原始事件信息进行自动化关联，最终可将这些疑似的攻击信息，关联分析形成精准的威胁告警事件。

四、XDR的架构和核心能力

1、XDR的架构

因为XDR是一个融合了多种安全检测、响应能力的平台框架，XDR的架构其关键组件包括前端组件（感应器，主要负责数据采集及检测）和后端平台组件（主要负责数据的汇聚、分析、威胁检测、响应处置）

XDR的架构
XDR前端组件，由生成安全遥测数据的“触角”（感应器）组成，这些触角包括但不限于EDR（终端检测与响应-Endpoint Detection and Response)、EPP(终端防护平台-Endpoint Protection Platforms)、NDR（流量检测与响应平台-Network Detection and Response)、SSE（安全服务边缘-Security Services Edge)、CWPP(云工作负载安全防护平台-Cloud Workload Protection Platforms)、蜜罐、邮件安全。

而XDR的后端平台，则是吸收所有关键位置的遥测数据、日志、威胁上下文信息，之后再对所有的数据进行关联、高级分析、从而完成威胁检测、调查分析、攻击溯源、工具编排、自动化响应等工作。

从整体架构上XDR可以看成是融合了各安全能力组件（EDR、NDR等）+SDC（安全数据中心）+SIEM/SA（安全信息事件管理/态势感知）+SOAR（安全编排与自动化响应）形成的一个安全运营系统。

2、核心能力

XDR核心能力包括数据集成、检测技术、可视化、编排响应技术。

  • 安全数据全面集成

对不同安全设备的数据进行全面的采集包括：内部（资产、脆弱性）、外部（流量、日志）以及云端威胁情报接入等相关安全数据的全面采集，汇聚、分析。

  • 安全威胁深度检测

对多源安全告警进行关联分析、规则分析、情报分析、机器学习等，发现潜伏的高级持续性威胁，提升告警检出率和准确率。

  • 安全态势集中展示

告警可视、事件可视、攻击可视，从多安全事件、攻击方向、攻击趋势、影响范围等多维度多视角进行态势可视化呈现。

  • 安全事件快速处置

通过可视化剧本编排，快速实现人员、流程、工具的有效协同，对接联动安全防护设备，在安全事件发生时自动下发阻断策略，并在必要时下发通知预警，及时完成安全闭环。

XDR成熟度模型如下：
XDR成熟度模型

五、XDR应用场景

XDR集中汇集云、网、端、威胁情报等多源安全数据/工具，解决安全运营过程中数据孤岛、告警疲劳、全局态势不可见、响应处置能力弱的问题。可以有效应用于企业常态化安全运营及重保攻防实战对抗场景。

XDR威胁检测联动响应示意图

  • 日常安全运营场景

日常安全运营，通过XDR提升安全运营的威胁检测能力和响应效率。
精准响应：更好的检测效果及响应能力，解决原有海量告警导致事件难以有效检测、溯源深度不足导致响应效果差的问题，有效应对攻防对抗加剧带来新的安全风险。
风险管理：从被动事件响应转向有计划的风险管理，预防重大安全事件。
处置闭环：采用简单有效的方式开展安全分析、管理、处置工作，实现事件处置的有效闭环管控。
态势可见：安全态势统一呈现，安全运营有地放矢，哪里不足补哪里。

  • 攻防实战对抗场景

攻防实战，通过XDR实现异构可扩展的威胁检测响应能力，进行快速响应、加固优化安全措施、攻击反制。
检测深度：更精准的高级威胁检测和安全事件溯源能力。
检测广度：拥有丰富的数据，包括事件完整的上下文信息、原始报文等供客户深度挖掘；全方位的的威胁数据采集，全面的威胁检测分析，全局的态势呈现。
敏捷响应：可进行快速响应、加固优化安全措施、攻击反制。

博客地址：http://xiejava.ishareread.com/

发表于 | 更新于: | 分类于 | | 阅读次数:
字数统计: 1.2k | 阅读时长 ≈ 4

实现了hexo的md文件格式解析，通过什么方式怎么来迁移hexo的博客到django的博客呢？开始想到的是通过Django的manage.py的shell命令，通过shell可以执行写好的python脚本进行hexo的md文件格式解析并入库。后来想想为啥不直接自定义一个manage.py的命令直接进行迁移呢？就如我们新建Django工程迁移数据库一样，执行python manage.py migrate来迁移数据库。我们可以定义python manage.py xxx来执行的迁移hexo博客。

自定义Django-admin命令分三步：创建management文件夹、编写命令代码、测试验证

一、创建management文件夹

自定义的Django-admin管理命令本质上是一个python脚本文件，它的存放路径必须遵循一定的规范，一般位于app/management/commands目录。整个文件夹的布局如下所示：
注意app要在setting中注册
在blog/management/commands包下面创建transblog.py文件
在这里插入图片描述

这里有两个要注意的地方:

1、app要在setting中注册。如blog在setting中注册了所以可以在blog/management/commands包下面创建transblog.py文件，api和common都没有在setting中注册所以在这两个目录下创建的不会作为管理命令生效。
2、创建的management/commands是Python包，不是单纯的目录，区别就是目录里必须有init.py文件，空文件都可以。

二、编写命令代码

创建命令管理文件后就可以在该文件中编写命令代码了。
每一个自定义的管理命令本质是一个Command类, 它继承了Django的Basecommand或其子类, 主要通过重写handle()方法实现自己的业务逻辑代码，而add_arguments()则用于帮助处理命令行的参数，如果运行命令时不需要额外参数，可以不写这个方法。
transblog.py的参考代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
# -*- coding: utf-8 -*-
"""
    :author: XieJava
    :url: http://ishareread.com
    :copyright: © 2021 XieJava <xiejava@ishareread.com>
    :license: MIT, see LICENSE for more details.
"""
import os

from blog.models import BlogPost,BlogCategory,Tag
from utils.parseblog import parseblog
from django.core.management.base import BaseCommand

class Command(BaseCommand):
    help='读取指定目录的.md文件解析至ishareblog'

    def add_arguments(self, parser):
        parser.add_argument('--path',help='输入.md文件的目录')
        parser.add_argument('--file',help='输入.md文件的路径')

    def handle(self, *args, **options):
        inputfile = input('请输入指定的目录路径或.md文件路径:')
        if inputfile.find('.md') > 0:
            self.transblogbyfile(inputfile)
        else:
            self.transblogbypath(inputfile)
        if options['path']:
            self.transblogbypath(options['path'])
        if options['file']:
            self.transblogbyfile(options['file'])

    '''读取目录解析md文件并写入数据库'''
    def transblogbypath(self,filepath='',):
        try:
            files = os.listdir(filepath)
            for file in files:
                if file.find('.md') > 0:
                    blog_file = os.path.join(filepath, file)
                    self.transblogbyfile(blog_file)
        except FileNotFoundError as e:
            print('请确认输入是否正确。',e)

    '''读取md文件入库'''
    def transblogbyfile(self,blogfile=''):
        blog_info = parseblog(blogfile)
        filename=os.path.basename(blogfile)
        #判断是否存在相同的title
        qs=BlogPost.objects.filter(title=blog_info['title'])
        if len(qs)==0:
            category = None
            tag_objs = []
            if 'categories' in blog_info:
                categories=blog_info['categories']
                if len(categories)>0:
                    category_title = categories[0]
                    category_qs=BlogCategory.objects.filter(title=category_title)
                    if len(category_qs)>0:
                        category=category_qs[0]
            if 'tags' in blog_info:
                tags=blog_info['tags']
                for tag in tags:
                    tag_obj,b=Tag.objects.get_or_create(tag=tag)
                    print(tag_obj.id)
                    tag_objs.append(tag_obj)
            blog = BlogPost()
            blog.title = blog_info['title']
            blog.content = blog_info['content']
            blog.isShow = 1 #默认显示
            blog.summary=blog.content[0:200] #默认提取内容的前200个字作为摘要
            blog.category=category
            blog.blogSource = filename
            blog.pubTime=blog_info['date']
            blog.save()
            if len(tag_objs)>0:
                blog.tags.add(*tag_objs)
            print(filename + '读取解析入库成功！')
        else:
            print(blog_info['title']+'已经存在！')

代码很简单，就是根据读取命令行的参数，这个参数就是需要迁移hexo的.md文件的目录或文件路径，读取目录或文件路径进行文件的解析，并写入到数据库。
.md文件的解析参考:Python二十行代码实现hexo的md文件格式解析

三、测试验证

命令代码写完后就可以进行测试了。
在命令行输入python manage.py 可以看到自定义的transblog已经加入到管理命令了
transblog已经加入到管理命令

输入 python manage.py transblog -h
会提示命令的参数和用法：
管理命令帮助

如执行 python manage.py transblog 会提示”请输入指定的目录路径或.md文件路径:”

1
2
(venv) PS D:\Python\study\pythonproject\ishareblog\ishareblog> python manage.py transblog
请输入指定的目录路径或.md文件路径:

我们输入需要迁移的.md文件或路径
如：E:\CloudStation\personal\xiejavablog\myhexo\myblog\source_posts\2022-07-27-Python二十行代码实现hexo的md文件格式解析.md
不出意外的情况下控制台会打印“XXXX.md读取解析入库成功！”的信息
执行效果

访问博客，可以看到文章已经迁移过来了
执行效果
全部代码仓库：https://gitee.com/ishareblog/ishareblog

作者博客：http://xiejava.ishareread.com/

发表于 | 更新于: | 分类于 | | 阅读次数:
字数统计: 521 | 阅读时长 ≈ 2

最近用django+Vue实现了一个博客应用，原来的hexo的博客用着也挺好，想继续留着用，于是就想将hexo生成的.md的博客内容文件解析后直接写到django的博客数据库里做同步显示。

hexo生成的.md文件内容主要分两部分，一部分是博客的信息、一部分是博客内容，博客信息包括标题、目录、标签、发布日期等，博客内容就是具体博客写的内容主体了。
hexo生成的.md文件内容

其中博客信息通过”—“来区分，夹在两个”—“块之间。博客的信息是yaml来描叙的需要解析并提取出相应的字段及内容，博客内容就更简单了直接是markdown描叙的不需要再解析了。
这里要做的事情就是提取两个”—“符号之间的内容，并解析相应的字段，提取两个”—“符号后面的内容作为博客的内容，形成字典，便于后面的入库。

代码示例如下：
yaml的解析可以直接用Python的PyYAML库

1
pip install PyYAML
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
# -*- coding: utf-8 -*-
"""
    :author: XieJava
    :url: http://ishareread.com
    :copyright: © 2021 XieJava <xiejava@ishareread.com>
    :license: MIT, see LICENSE for more details.
"""
import yaml

'''将md文件转成blog对象'''
def parseblog(blog_md_file):
    #读md文件
    md_f = open(blog_md_file, "r",encoding='utf-8')
    md_f_str=md_f.read()
    #解析两个---之间的内容
    pattern='---'
    blog_data={}
    pattern_list=list(pattern_search(md_f_str, pattern))
    if len(pattern_list)>=2:
        blog_info_str=md_f_str[pattern_list[0]+len(pattern):pattern_list[1]]
        blog_data=yaml.load(blog_info_str,Loader=yaml.SafeLoader)
        blog_data['content']=md_f_str[pattern_list[1]+len(pattern):]
    md_f.close()
    return blog_data

'''分割符号匹配检索'''
def pattern_search(string,pattern):
    index=0
    while index<len(string)-len(pattern):
        index=string.find(pattern,index,len(string))
        if index==-1:
            break
        yield index
        index+=len(pattern)-1

if __name__ == '__main__':
    blog_data=parseblog('E:\\CloudStation\\personal\\xiejavablog\\myhexo\\myblog\\source\\_posts\\2022-07-19-Vue3解析markdown解析并实现代码高亮显示.md')
    print(blog_data)

效果：
可以看到hexo的md文件正确解析出来，形成字典。后续可以直接进行入库操作了。
解析效果
Python的库还是很丰富实用的，用Python二十行代码就实现hexo的md文件格式解析。

源代码仓库：https://gitee.com/ishareblog/ishareblog

发表于 | 更新于: | 分类于 ， | | 阅读次数:
字数统计: 2.4k | 阅读时长 ≈ 12

Django是一个开放源代码的Web应用框架，由Python写成。采用了MTV的框架模式，即模型M，视图V和模版T。它最初是被开发来用于管理劳伦斯出版集团旗下的一些以新闻内容为主的网站的，即是CMS（内容管理系统）软件。对于博客网站来说是典型的CMS应用。本文介绍通过Django+Vue的博客模版快速实现一个可用的博客网站。

这里用的博客模板是Gblog是一款nice的基于 vue 的博客模板。界面简洁轻快，非常适合用作个人博客。https://gitee.com/fengziy/Gblog 后台的接口和管理界面就通过Django框架来实现了。

这里数据库用mysql，接口框架主要用到的是Django的djangorestframework，内容编辑器用的是markdown通过django-mdedior库实现。

一、依赖库

1
2
3
4
5
6
7
8
9
10
11
asgiref==3.5.2
Django==4.0.6
django-cors-headers==3.13.0
django-filter==22.1
django-mdeditor==0.1.20
djangorestframework==3.13.1
mysqlclient==2.1.1
Pillow==9.2.0
pytz==2022.1
sqlparse==0.4.2
tzdata==2022.1

二、工程目录组织结构

工程目录组织结构

三、代码实现

1、模型

模型很简单，根据Gblog前台要显示的内容包括有‘文章分类’、‘文章标签’、‘博客文章’、‘站点信息’、‘社交信息’、‘聚焦’，模型定义分别如下：
这里要说明的是因为博客文章内容准备用markdown编写，所以引入了mdeditor from mdeditor.fields import MDTextField
内容字段content=MDTextField(verbose_name='内容')
模型代码示例如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
from django.db import models
from common.basemodel import BaseModel
from mdeditor.fields import MDTextField
# Create your models here.
'''文章分类'''
class BlogCategory(BaseModel):
    id = models.AutoField(primary_key=True)
    title=models.CharField(max_length=50,verbose_name='分类名称',default='')
    href=models.CharField(max_length=100,verbose_name='分类路径',default='')

    def __str__(self):
        return self.title

    class Meta:
        verbose_name='文章分类'
        verbose_name_plural='文章分类'


'''文章标签'''
class Tag(BaseModel):
    id=models.AutoField(primary_key=True)
    tag=models.CharField(max_length=20, verbose_name='标签')

    def __str__(self):
        return self.tag

    class Meta:
        verbose_name='标签'
        verbose_name_plural='标签'


'''博客文章'''
class BlogPost(BaseModel):
    id = models.AutoField(primary_key=True)
    title = models.CharField(max_length=200, verbose_name='文章标题', unique = True)
    category = models.ForeignKey(BlogCategory, blank=True, null=True, verbose_name='文章分类', on_delete=models.DO_NOTHING)
    isTop=models.BooleanField(default=False,verbose_name='是否置顶')
    isHot=models.BooleanField(default=False,verbose_name='是否热门')
    summary=models.CharField(max_length=500,verbose_name='内容摘要',default='')
    content=MDTextField(verbose_name='内容')
    viewsCount= models.IntegerField(default=0, verbose_name="查看数")
    commentsCount=models.IntegerField(default=0, verbose_name="评论数")
    tags=models.ManyToManyField(to=Tag, related_name="tag_post", blank=True, default=None,verbose_name="标签")


    @property
    def tag_list(self):
        return ','.join([i.tag for i in self.tags.all()])

    def __str__(self):
            return self.title

    class Meta:
        verbose_name = '博客文章'
        verbose_name_plural = '博客文章'


'''站点信息'''
class Site(BaseModel):
    id = models.AutoField(primary_key=True)
    name=models.CharField(max_length=50, verbose_name='站点名称', unique = True)
    avatar=models.CharField(max_length=200, verbose_name='站点图标')
    slogan=models.CharField(max_length=200, verbose_name='站点标语')
    domain=models.CharField(max_length=200, verbose_name='站点域名')
    notice=models.CharField(max_length=200, verbose_name='站点备注')
    desc=models.CharField(max_length=200, verbose_name='站点描述')

    def __str__(self):
            return self.name

    class Meta:
        verbose_name = '站点信息'
        verbose_name_plural = '站点信息'


'''社交信息'''
class Social(BaseModel):
    id=models.AutoField(primary_key=True)
    title=models.CharField(max_length=20, verbose_name='标题')
    icon=models.CharField(max_length=200, verbose_name='图标')
    color=models.CharField(max_length=20, verbose_name='颜色')
    href=models.CharField(max_length=100, verbose_name='路径')

    def __str__(self):
        return self.title

    class Meta:
        verbose_name = '社交信息'
        verbose_name_plural = '社交信息'

'''聚焦'''
class Focus(BaseModel):
    id=models.AutoField(primary_key=True)
    title=models.CharField(max_length=20, verbose_name='标题')
    img=models.CharField(max_length=100, verbose_name='路径')

    def __str__(self):
        return self.title

    class Meta:
        verbose_name='聚焦'
        verbose_name_plural='聚焦'

'''友链'''
class Friend(BaseModel):
    id=models.AutoField(primary_key=True)
    siteName=models.CharField(max_length=20, verbose_name='友链站点名称')
    path=models.CharField(max_length=100, verbose_name='地址路径')
    desc=models.CharField(max_length=200, verbose_name='描述')

    def __str__(self):
        return self.siteName

    class Meta:
        verbose_name='友链'
        verbose_name_plural='友链'

2、admin管理

实际上只要把模型注册到admin就可以了

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from django.contrib import admin
from blog.models import *
# Register your models here.
@admin.register(BlogCategory)
class BlogCategoryAdmin(admin.ModelAdmin):
    admin.site.site_title="ishareblog后台"
    admin.site.site_header="ishareblog后台"
    admin.site.index_title="ishareblog管理"

    list_display = ['id', 'title', 'href']

@admin.register(BlogPost)
class BlogPostAdmin(admin.ModelAdmin):
    list_display = ['title','category','isTop','isHot']
    search_fields = ('title',)

@admin.register(Site)
class SiteAdmin(admin.ModelAdmin):
    list_display = ['name','slogan','domain','desc']

@admin.register(Social)
class SocialAdmin(admin.ModelAdmin):
    list_display = ['title','href']

@admin.register(Focus)
class FoucusAdmin(admin.ModelAdmin):
    list_display = ['title','img']

@admin.register(Friend)
class FoucusAdmin(admin.ModelAdmin):
    list_display = ['siteName','path','desc']

@admin.register(Tag)
class TagAdmin(admin.ModelAdmin):
    list_display = ['id','tag']

3、接口

前端是Vue模板展示的，所以要为前端Vue提供相应的接口。通过djangorestframework将模型通过restful接口提供是非常easy的。

1）首先将需要暴露的模型通过序列化类序列化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
serializers.py

from blog.models import *
from rest_framework import serializers
class BlogCategoryModelSerializer(serializers.ModelSerializer):
    class Meta:
        model=BlogCategory
        fields = "__all__"

class TagModelSerializer(serializers.ModelSerializer):
    class Meta:
        model = Tag
        fields = "__all__"


class BlogPostModelSerializer(serializers.ModelSerializer):
    create_time = serializers.DateTimeField(format="%Y-%m-%d %H:%M:%S", required=False, read_only=True)
    update_time = serializers.DateTimeField(format="%Y-%m-%d %H:%M:%S", required=False, read_only=True)
    category_id = serializers.CharField(max_length=32, source='category.id')
    pubTime=update_time
    category=BlogCategoryModelSerializer()
    tags=serializers.SerializerMethodField()

    # 多对多，钩子函数序列化,必须是以get_开头的
    def get_tags(self, obj):
        tags = obj.tags.all()
        tag = TagModelSerializer(tags, many=True)
        return tag.data

    class Meta:
        model=BlogPost
        fields="__all__"

class SiteModelSerializer(serializers.ModelSerializer):
    class Meta:
        model = Site
        fields = "__all__"

class SocialModelSerializer(serializers.ModelSerializer):
    class Meta:
        model = Social
        fields = "__all__"

class FocusModelSerializer(serializers.ModelSerializer):
    class Meta:
        model = Focus
        fields = "__all__"

class FriendModelSerializer(serializers.ModelSerializer):
    class Meta:
        model = Friend
        fields = "__all__"

2）将序列化的对象通过视图类提供接口

custommodelviewset.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
from rest_framework import status
from rest_framework import viewsets
from common.customresponse import CustomResponse

class CustomModelViewSet(viewsets.ModelViewSet):

    #CreateModelMixin->create
    def create(self, request, *args, **kwargs):
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        self.perform_create(serializer)
        headers = self.get_success_headers(serializer.data)
        return CustomResponse(data=serializer.data, code=201,msg="OK", status=status.HTTP_201_CREATED,headers=headers)
    #ListModelMixin->list
    def list(self, request, *args, **kwargs):
        queryset = self.filter_queryset(self.get_queryset())
        page = self.paginate_queryset(queryset)
        if page is not None:
            serializer = self.get_serializer(page, many=True)
            return self.get_paginated_response(serializer.data)

        serializer = self.get_serializer(queryset, many=True)
        return CustomResponse(data=serializer.data, code=200, msg="OK", status=status.HTTP_200_OK)

    #RetrieveModelMixin->retrieve
    def retrieve(self, request, *args, **kwargs):
        instance = self.get_object()
        serializer = self.get_serializer(instance)
        return CustomResponse(data=serializer.data, code=200, msg="OK", status=status.HTTP_200_OK)
    #UpdateModelMixin->update
    def update(self, request, *args, **kwargs):
        partial = kwargs.pop('partial', False)
        instance = self.get_object()
        serializer = self.get_serializer(instance, data=request.data, partial=partial)
        serializer.is_valid(raise_exception=True)
        self.perform_update(serializer)

        if getattr(instance, '_prefetched_objects_cache', None):
            # If 'prefetch_related' has been applied to a queryset, we need to
            # forcibly invalidate the prefetch cache on the instance.
            instance._prefetched_objects_cache = {}

        return CustomResponse(data=serializer.data, code=200, msg="OK", status=status.HTTP_200_OK)

    #DestroyModelMixin->destroy
    def destroy(self, request, *args, **kwargs):
        instance = self.get_object()
        self.perform_destroy(instance)
        return CustomResponse(data=[], code=204, msg="OK", status=status.HTTP_204_NO_CONTENT)

views.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
# Create your views here.
from django_filters.rest_framework import DjangoFilterBackend
from rest_framework import viewsets, status
from rest_framework import filters
from api.myfilter import BlogPostFilter
from api.serializers import *
from blog.models import BlogCategory, BlogPost,Site,Social,Focus,Friend,Tag
from api.mypage import MyPage
from common.custommodelviewset import CustomModelViewSet
from common.customresponse import CustomResponse

class BlogCategoryViewset(CustomModelViewSet):
    queryset = BlogCategory.objects.all()
    serializer_class = BlogCategoryModelSerializer

class BlogsView(CustomModelViewSet):
    queryset = BlogPost.objects.order_by('-isTop','-update_time')
    serializer_class = BlogPostModelSerializer
    pagination_class = MyPage
    filter_backends = (DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter)
    filterset_class = BlogPostFilter
    #搜索
    search_fields=('title',)
    #排序
    ordering_fields = ('isTop', 'update_time')
    #自定义获取详情接口
    def retrieve(self,request,*args, **kwargs):
        instance=self.get_object()
        instance.viewsCount+=1
        instance.save()
        serializer=self.get_serializer(instance)
        return CustomResponse(data=serializer.data,code=200,msg="success",status=status.HTTP_200_OK)


class SiteView(CustomModelViewSet):
    queryset = Site.objects.all()
    serializer_class = SiteModelSerializer

class SocialView(CustomModelViewSet):
    queryset = Social.objects.all()
    serializer_class = SocialModelSerializer

class FocusView(CustomModelViewSet):
    queryset = Focus.objects.all()
    serializer_class = FocusModelSerializer

class FriendView(viewsets.ModelViewSet):
    queryset = Friend.objects.all()
    serializer_class = FriendModelSerializer

class TagView(viewsets.ModelViewSet):
    queryset = Tag.objects.all()
    serializer_class = TagModelSerializer

3）通过路由来实现接口地址和视图的绑定和访问

urls.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
# -*- coding: utf-8 -*-
"""
    :author: XieJava
    :url: http://ishareread.com
    :copyright: © 2021 XieJava <xiejava@ishareread.com>
    :license: MIT, see LICENSE for more details.
"""
from api import views
from django.urls import path,include
from rest_framework.routers import DefaultRouter
blogcategory_list=views.BlogCategoryViewset.as_view({'get':'list',})
blogcategory_detail=views.BlogCategoryViewset.as_view({ 'get': 'retrieve',})
blog_list=views.BlogsView.as_view({'get':'list',})
blog_detail=views.BlogsView.as_view({ 'get': 'retrieve',})
site_list=views.SiteView.as_view({'get':'list',})
site_detail=views.SiteView.as_view({'get':'retrieve',})
social_list=views.SocialView.as_view({'get':'list',})
social_detail=views.SocialView.as_view({'get':'retrieve',})
focus_list=views.FocusView.as_view({'get':'list',})
focus_detail=views.FocusView.as_view({'get':'retrieve'})
friend_list=views.FriendView.as_view({'get':'list',})
friend_detail=views.FriendView.as_view({'get':'retrieve'})
tags_list=views.TagView.as_view({'get':'list',})
# router=DefaultRouter()
# router.register('blogs',views.BlogsView)
urlpatterns = [
    path('category/',blogcategory_list),
    path('category/<pk>/',blogcategory_detail),
    path('post/list',blog_list),
    path('post/<pk>',blog_detail),
    path('social/',social_list),
    path('site/<pk>',site_detail),
    path('focus/list',focus_list),
    path('comment/',blog_list),
    path('friend/',friend_list),
    path('tags/',tags_list),
]

4）自定义接口返回格式

接口需要根据Glog定义的格式进行定义和返回，这里就需要自定义接口返回格式。
具体实现参见：https://xiejava.blog.csdn.net/article/details/125773730
–自定义返回响应类customresponse.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from rest_framework.response import Response
from rest_framework.serializers import Serializer

class CustomResponse(Response):
    def __init__(self,data=None,code=None,msg=None,
                 status=None,
                 template_name=None, headers=None,
                 exception=False, content_type=None,**kwargs):
        super().__init__(None, status=status)

        if isinstance(data, Serializer):
            msg = (
                'You passed a Serializer instance as data, but '
                'probably meant to pass serialized `.data` or '
                '`.error`. representation.'
            )
            raise AssertionError(msg)
        self.data={'code':code,'msg':msg,'data':data}
        self.data.update(kwargs)
        self.template_name=template_name
        self.exception=exception
        self.content_type=content_type

        if headers:
            for name, value in headers.items():
                self[name] = value

–翻页实现类mypage.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from rest_framework import status
from rest_framework.pagination import PageNumberPagination
from common.customresponse import CustomResponse

class MyPage(PageNumberPagination):
    page_size = 8 #每页显示数量
    max_page_size = 50 #每页最大显示数量。
    page_size_query_param = 'size' #每页数量的参数名称
    page_query_param = 'page'  #页码的参数名称

    #自定义分页器的返回参数
    def get_paginated_response(self, data):
        ret_data = dict()
        ret_data['items'] = data
        # 加入自定义分页信息
        ret_data['total'] = self.page.paginator.count
        ret_data['hasNextPage'] = self.get_next_link()
        ret_data['size'] = self.page_size
        ret_data['page'] = self.page.number
        return CustomResponse(data=ret_data,code=200,msg="OK",status=status.HTTP_200_OK)

全部代码：
后台代码：https://gitee.com/xiejava/ishareblog
前台代码：https://gitee.com/xiejava/Gblog

四、效果

1、后台管理

管理界面
管理界面
博客文章列表
博客文章列表
文章内容编辑，支持markdown
文章内容编辑，支持markdown
分类管理
文章分类
标签管理
标签管理
社交信息
社交信息

2、接口

接口清单
接口清单
文章列表接口，支持翻页
文章列表接口

文章详情接口
文章详情接口

3、前台展现

前台展现

文章列表
文章列表
文章详情，支持markdown显示及目录
文章详情

社交信息
社交信息

博客效果地址：http://blog.ishareread.com

后续考虑
1、django原生admin的管理界面还是简陋了一点，后续可能会用其他管理界面的UI给换掉
2、现在有了一个hexo的博客了，后续可能会考虑实现hexo生成的博客内容直接同步到django的博客，或者django博客编辑的内容直接生成hexo的.md文件
有兴趣的话可以关注本博客

博客：http://xiejava.ishareread.com

发表于 | 更新于: | 分类于 ， | | 阅读次数:
字数统计: 466 | 阅读时长 ≈ 2

Vue实现博客前端，需要实现markdown的解析，如果有代码则需要实现代码的高亮。
Vue的markdown解析库有很多，如markdown-it、vue-markdown-loader、marked、vue-markdown等。这些库都大同小异。这里选用的是marked，代码高亮的库选用的是highlight.js。

具体实现步骤如下：

一、安装依赖库

在vue项目下打开命令窗口，并输入以下命令

1
2
npm install marked -save    // marked 用于将markdown转换成html
npm install highlight.js -save   //用于代码高亮显示

命令执行完后可以在控制台或package.json文件中看到有安装的版本号
package.json文件中看到有安装的版本号

二、在main.js文件中引入highlight.js及样式并创建一个自定义的全局指令

1
2
3
4
5
6
7
8
9
10
import hljs from 'highlight.js';
import 'highlight.js/styles/atom-one-dark.css' //样式

//创建v-highlight全局指令
Vue.directive('highlight',function (el) {
  let blocks = el.querySelectorAll('pre code');
  blocks.forEach((block)=>{
    hljs.highlightBlock(block)
  })
})

这样就可以在vue组件中使用v-highlight引用代码高亮的方法了。

三、在Vue组件中应用marked解析及实现代码高亮

代码示例如下：

1
2
3
4
 <!-- 正文输出 -->
<div class="entry-content">
  <div v-highlight v-html="article"  id="content"></div>
</div>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
<script>
    // 将marked 引入
  import { marked }from 'marked';
    export default {
        name: 'articles',
        data(){
          return{
              article:''
          }
        },
        methods: {
          getPostDetail() {
            console.log('getPostDetail()'+this.id)
            fetchPostDetail(this.id).then(res => {
               this.postdetail=res.data
               // 调用marked()方法，将markdown转换成html
               this.article= marked(this.postdetail.content);
               console.log(res.data)
              }).catch(err => {
                  console.log(err)
              })

          },
        created() {
          //调用获取文章内容的接口方法
          this.getPostDetail()
        },
    }
 </script>

四、显示效果

markdown解析及代码高亮显示效果
在这里插入图片描述

示例中引用的样式是 import 'highlight.js/styles/atom-one-dark.css'
实际highlight.js/styles中提供了很多样式，可以根据自己的喜好选用。

代码高亮样式

博客：http://xiejava.ishareread.com/

0%