|
|
# 导入Python内置time模块，用于生成唯一ID（时间戳毫秒级）
|
|
|
import time
|
|
|
|
|
|
# 导入Elasticsearch客户端模块，用于直接操作Elasticsearch服务（如创建管道、删除索引）
|
|
|
import elasticsearch.client
|
|
|
# 导入Django配置模块，用于读取项目中的Elasticsearch配置（settings.py中）
|
|
|
from django.conf import settings
|
|
|
# 从elasticsearch-dsl库导入核心组件：
|
|
|
# Document：Elasticsearch文档模型基类（类似Django的Model）
|
|
|
# InnerDoc：嵌套文档基类（用于存储结构化子数据，如地理位置、用户代理信息）
|
|
|
# 字段类型：Date(日期)、Integer(整数)、Long(长整数)、Text(可分词文本)、Object(对象类型)、GeoPoint(地理坐标)、Keyword(不可分词文本)、Boolean(布尔值)
|
|
|
from elasticsearch_dsl import Document, InnerDoc, Date, Integer, Long, Text, Object, GeoPoint, Keyword, Boolean
|
|
|
# 导入elasticsearch-dsl的连接管理模块，用于建立与Elasticsearch服务的连接
|
|
|
from elasticsearch_dsl.connections import connections
|
|
|
|
|
|
# 从当前应用（blog）的models.py导入Article模型，用于将文章数据同步到Elasticsearch
|
|
|
from blog.models import Article
|
|
|
|
|
|
# 判断项目是否启用Elasticsearch：检查settings.py中是否配置了ELASTICSEARCH_DSL
|
|
|
ELASTICSEARCH_ENABLED = hasattr(settings, 'ELASTICSEARCH_DSL')
|
|
|
|
|
|
# 如果启用了Elasticsearch，执行以下初始化操作
|
|
|
if ELASTICSEARCH_ENABLED:
|
|
|
# 建立与Elasticsearch服务的连接：从settings中读取配置的主机地址（如['http://localhost:9200']）
|
|
|
connections.create_connection(
|
|
|
hosts=[settings.ELASTICSEARCH_DSL['default']['hosts']])
|
|
|
# 导入Elasticsearch原生客户端，用于执行更底层的操作（如创建索引、删除索引）
|
|
|
from elasticsearch import Elasticsearch
|
|
|
|
|
|
# 初始化Elasticsearch原生客户端，传入服务地址
|
|
|
es = Elasticsearch(settings.ELASTICSEARCH_DSL['default']['hosts'])
|
|
|
# 导入Elasticsearch的IngestClient（数据处理管道客户端），用于创建数据预处理管道
|
|
|
from elasticsearch.client import IngestClient
|
|
|
|
|
|
# 初始化IngestClient，绑定到上面创建的Elasticsearch客户端
|
|
|
c = IngestClient(es)
|
|
|
try:
|
|
|
# 尝试获取名为'geoip'的数据处理管道（用于解析IP地址对应的地理位置）
|
|
|
c.get_pipeline('geoip')
|
|
|
# 如果管道不存在（捕获NotFoundError异常），则创建该管道
|
|
|
except elasticsearch.exceptions.NotFoundError:
|
|
|
# 创建'geoip'管道：定义数据处理逻辑，通过geoip处理器解析IP地址
|
|
|
c.put_pipeline('geoip', body='''{
|
|
|
"description" : "Add geoip info", // 管道描述：添加地理位置信息
|
|
|
"processors" : [ // 处理器列表：定义数据处理步骤
|
|
|
{
|
|
|
"geoip" : { // geoip处理器：Elasticsearch内置，用于解析IP
|
|
|
"field" : "ip" // 待解析的字段：文档中的'ip'字段
|
|
|
}
|
|
|
}
|
|
|
]
|
|
|
}''')
|
|
|
|
|
|
|
|
|
# 定义GeoIp嵌套文档类（InnerDoc）：存储IP地址解析后的地理位置信息
|
|
|
class GeoIp(InnerDoc):
|
|
|
continent_name = Keyword() # 洲名（Keyword类型：不可分词，适合精确查询/排序）
|
|
|
country_iso_code = Keyword() # 国家ISO代码（如CN、US，Keyword类型）
|
|
|
country_name = Keyword() # 国家名称（Keyword类型）
|
|
|
location = GeoPoint() # 地理坐标（经纬度，GeoPoint类型：支持地理位置查询）
|
|
|
|
|
|
|
|
|
# 定义UserAgentBrowser嵌套文档类：存储用户代理（UA）中的浏览器信息
|
|
|
class UserAgentBrowser(InnerDoc):
|
|
|
Family = Keyword() # 浏览器家族（如Chrome、Firefox，Keyword类型）
|
|
|
Version = Keyword() # 浏览器版本（如120.0，Keyword类型）
|
|
|
|
|
|
|
|
|
# 定义UserAgentOS嵌套文档类：存储用户代理中的操作系统信息，继承自UserAgentBrowser（结构一致）
|
|
|
class UserAgentOS(UserAgentBrowser):
|
|
|
pass # 直接继承父类字段，无需额外定义
|
|
|
|
|
|
|
|
|
# 定义UserAgentDevice嵌套文档类：存储用户代理中的设备信息
|
|
|
class UserAgentDevice(InnerDoc):
|
|
|
Family = Keyword() # 设备家族（如iPhone、Windows，Keyword类型）
|
|
|
Brand = Keyword() # 设备品牌（如Apple、Huawei，Keyword类型）
|
|
|
Model = Keyword() # 设备型号（如iPhone 15，Keyword类型）
|
|
|
|
|
|
|
|
|
# 定义UserAgent嵌套文档类：存储完整的用户代理信息（包含浏览器、OS、设备）
|
|
|
class UserAgent(InnerDoc):
|
|
|
browser = Object(UserAgentBrowser, required=False) # 浏览器信息（Object类型：关联UserAgentBrowser）
|
|
|
os = Object(UserAgentOS, required=False) # 操作系统信息（Object类型：关联UserAgentOS）
|
|
|
device = Object(UserAgentDevice, required=False) # 设备信息（Object类型：关联UserAgentDevice）
|
|
|
string = Text() # 完整UA字符串（Text类型：可分词，支持模糊查询）
|
|
|
is_bot = Boolean() # 是否为爬虫（Boolean类型：true/false）
|
|
|
|
|
|
|
|
|
# 定义ElapsedTimeDocument文档类：Elasticsearch中的"性能监控"文档模型（记录请求耗时、访问信息）
|
|
|
class ElapsedTimeDocument(Document):
|
|
|
url = Keyword() # 访问URL（Keyword类型：精确匹配，不分词）
|
|
|
time_taken = Long() # 请求耗时（毫秒，Long类型：支持大范围数值存储）
|
|
|
log_datetime = Date() # 日志记录时间（Date类型：支持时间范围查询）
|
|
|
ip = Keyword() # 访问IP地址（Keyword类型：精确匹配）
|
|
|
geoip = Object(GeoIp, required=False) # 地理位置信息（Object类型：关联GeoIp嵌套文档，非必填）
|
|
|
useragent = Object(UserAgent, required=False) # 用户代理信息（Object类型：关联UserAgent嵌套文档，非必填）
|
|
|
|
|
|
# 定义文档对应的Elasticsearch索引配置
|
|
|
class Index:
|
|
|
name = 'performance' # 索引名称：Elasticsearch中存储性能数据的索引名
|
|
|
settings = { # 索引设置
|
|
|
"number_of_shards": 1, # 分片数：1个（小型索引无需多分片）
|
|
|
"number_of_replicas": 0 # 副本数：0个（开发/小型场景无需副本，节省资源）
|
|
|
}
|
|
|
|
|
|
# 定义文档元数据（兼容Elasticsearch旧版本，doc_type在7.x后已废弃，此处保留兼容）
|
|
|
class Meta:
|
|
|
doc_type = 'ElapsedTime' # 文档类型：标识索引中的文档类别
|
|
|
|
|
|
|
|
|
# 定义ElaspedTimeDocumentManager类：ElapsedTimeDocument的管理类（封装索引创建、数据插入等操作）
|
|
|
class ElaspedTimeDocumentManager:
|
|
|
# 静态方法：创建性能监控索引（如果不存在）
|
|
|
@staticmethod
|
|
|
def build_index():
|
|
|
# 导入Elasticsearch原生客户端
|
|
|
from elasticsearch import Elasticsearch
|
|
|
# 初始化客户端，读取settings中的Elasticsearch地址
|
|
|
client = Elasticsearch(settings.ELASTICSEARCH_DSL['default']['hosts'])
|
|
|
# 检查名为'performance'的索引是否已存在
|
|
|
res = client.indices.exists(index="performance")
|
|
|
# 如果索引不存在，初始化ElapsedTimeDocument（创建索引及映射）
|
|
|
if not res:
|
|
|
ElapsedTimeDocument.init()
|
|
|
|
|
|
# 静态方法：删除性能监控索引
|
|
|
@staticmethod
|
|
|
def delete_index():
|
|
|
from elasticsearch import Elasticsearch
|
|
|
es = Elasticsearch(settings.ELASTICSEARCH_DSL['default']['hosts'])
|
|
|
# 删除'performance'索引，忽略400（请求错误）和404（索引不存在）异常
|
|
|
es.indices.delete(index='performance', ignore=[400, 404])
|
|
|
|
|
|
# 静态方法：创建性能监控文档（插入一条访问耗时记录）
|
|
|
@staticmethod
|
|
|
def create(url, time_taken, log_datetime, useragent, ip):
|
|
|
# 确保索引已创建（调用build_index方法）
|
|
|
ElaspedTimeDocumentManager.build_index()
|
|
|
# 初始化UserAgent嵌套文档对象
|
|
|
ua = UserAgent()
|
|
|
# 赋值浏览器信息：从传入的useragent对象中提取浏览器家族和版本
|
|
|
ua.browser = UserAgentBrowser()
|
|
|
ua.browser.Family = useragent.browser.family
|
|
|
ua.browser.Version = useragent.browser.version_string
|
|
|
|
|
|
# 赋值操作系统信息：从传入的useragent对象中提取OS家族和版本
|
|
|
ua.os = UserAgentOS()
|
|
|
ua.os.Family = useragent.os.family
|
|
|
ua.os.Version = useragent.os.version_string
|
|
|
|
|
|
# 赋值设备信息：从传入的useragent对象中提取设备家族、品牌、型号
|
|
|
ua.device = UserAgentDevice()
|
|
|
ua.device.Family = useragent.device.family
|
|
|
ua.device.Brand = useragent.device.brand
|
|
|
ua.device.Model = useragent.device.model
|
|
|
# 赋值完整UA字符串和是否为爬虫的标识
|
|
|
ua.string = useragent.ua_string
|
|
|
ua.is_bot = useragent.is_bot
|
|
|
|
|
|
# 初始化ElapsedTimeDocument文档对象，设置字段值
|
|
|
doc = ElapsedTimeDocument(
|
|
|
meta={
|
|
|
'id': int(round(time.time() * 1000)) # 文档ID：毫秒级时间戳（确保唯一）
|
|
|
},
|
|
|
url=url, # 访问URL
|
|
|
time_taken=time_taken, # 请求耗时（毫秒）
|
|
|
log_datetime=log_datetime,# 记录时间
|
|
|
useragent=ua, # 用户代理信息（嵌套文档）
|
|
|
ip=ip) # 访问IP
|
|
|
# 保存文档到Elasticsearch，并指定使用'geoip'管道预处理（解析IP地址）
|
|
|
doc.save(pipeline="geoip")
|
|
|
|
|
|
|
|
|
# 定义ArticleDocument文档类：Elasticsearch中的"文章"文档模型（用于文章搜索）
|
|
|
class ArticleDocument(Document):
|
|
|
# 文章内容：Text类型，使用ik_max_word分词器（分词更细，适合全文搜索），搜索时用ik_smart（分词更粗，提升效率）
|
|
|
body = Text(analyzer='ik_max_word', search_analyzer='ik_smart')
|
|
|
# 文章标题：同上，支持中文分词搜索
|
|
|
title = Text(analyzer='ik_max_word', search_analyzer='ik_smart')
|
|
|
# 作者信息：Object类型，包含昵称（可分词）和ID（整数）
|
|
|
author = Object(properties={
|
|
|
'nickname': Text(analyzer='ik_max_word', search_analyzer='ik_smart'),
|
|
|
'id': Integer()
|
|
|
})
|
|
|
# 分类信息：Object类型，包含分类名称（可分词）和ID（整数）
|
|
|
category = Object(properties={
|
|
|
'name': Text(analyzer='ik_max_word', search_analyzer='ik_smart'),
|
|
|
'id': Integer()
|
|
|
})
|
|
|
# 标签信息：Object类型（数组），每个标签包含名称（可分词）和ID（整数）
|
|
|
tags = Object(properties={
|
|
|
'name': Text(analyzer='ik_max_word', search_analyzer='ik_smart'),
|
|
|
'id': Integer()
|
|
|
})
|
|
|
|
|
|
pub_time = Date() # 发布时间（Date类型：支持按时间排序/筛选）
|
|
|
status = Text() # 文章状态（如'p'=发布，'d'=草稿，Text类型）
|
|
|
comment_status = Text() # 评论状态（如'o'=开启，'c'=关闭，Text类型）
|
|
|
type = Text() # 文章类型（如'p'=页面，'a'=普通文章，Text类型）
|
|
|
views = Integer() # 浏览量（Integer类型：支持数值排序）
|
|
|
article_order = Integer() # 排序权重（Integer类型：用于自定义文章排序）
|
|
|
|
|
|
# 定义文档对应的Elasticsearch索引配置
|
|
|
class Index:
|
|
|
name = 'blog' # 索引名称：存储文章数据的索引名
|
|
|
settings = { # 索引设置
|
|
|
"number_of_shards": 1, # 分片数：1个（小型博客无需多分片）
|
|
|
"number_of_replicas": 0 # 副本数：0个（开发/小型场景节省资源）
|
|
|
}
|
|
|
|
|
|
# 文档元数据（兼容旧版本Elasticsearch的doc_type）
|
|
|
class Meta:
|
|
|
doc_type = 'Article' # 文档类型：标识为文章类文档
|
|
|
|
|
|
|
|
|
# 定义ArticleDocumentManager类：ArticleDocument的管理类（封装文章索引的创建、重建、更新等操作）
|
|
|
class ArticleDocumentManager():
|
|
|
# 构造方法：实例化管理类时自动创建文章索引（如果不存在）
|
|
|
def __init__(self):
|
|
|
self.create_index()
|
|
|
|
|
|
# 实例方法：创建文章索引（调用ArticleDocument的init方法，生成索引和字段映射）
|
|
|
def create_index(self):
|
|
|
ArticleDocument.init()
|
|
|
|
|
|
# 实例方法：删除文章索引
|
|
|
def delete_index(self):
|
|
|
from elasticsearch import Elasticsearch
|
|
|
es = Elasticsearch(settings.ELASTICSEARCH_DSL['default']['hosts'])
|
|
|
# 删除'blog'索引，忽略400和404异常
|
|
|
es.indices.delete(index='blog', ignore=[400, 404])
|
|
|
|
|
|
# 实例方法：将Django的Article模型对象列表转换为Elasticsearch的ArticleDocument列表
|
|
|
def convert_to_doc(self, articles):
|
|
|
# 列表推导式：遍历每篇文章，构建对应的ArticleDocument
|
|
|
return [
|
|
|
ArticleDocument(
|
|
|
meta={'id': article.id}, # 文档ID与Django Article模型ID一致（便于关联）
|
|
|
body=article.body, # 文章内容
|
|
|
title=article.title, # 文章标题
|
|
|
author={ # 作者信息：从Article模型的author字段提取
|
|
|
'nickname': article.author.username,
|
|
|
'id': article.author.id
|
|
|
},
|
|
|
category={ # 分类信息：从Article模型的category字段提取
|
|
|
'name': article.category.name,
|
|
|
'id': article.category.id
|
|
|
},
|
|
|
tags=[ # 标签信息：遍历Article模型的tags多对多字段，提取每个标签的名称和ID
|
|
|
{'name': t.name, 'id': t.id} for t in article.tags.all()
|
|
|
],
|
|
|
pub_time=article.pub_time, # 发布时间
|
|
|
status=article.status, # 文章状态
|
|
|
comment_status=article.comment_status, # 评论状态
|
|
|
type=article.type, # 文章类型
|
|
|
views=article.views, # 浏览量
|
|
|
article_order=article.article_order # 排序权重
|
|
|
) for article in articles]
|
|
|
|
|
|
# 实例方法：重建文章索引（全量同步文章数据到Elasticsearch）
|
|
|
def rebuild(self, articles=None):
|
|
|
# 确保索引已创建（初始化索引和映射）
|
|
|
ArticleDocument.init()
|
|
|
# 如果传入了articles参数，则同步指定文章；否则同步所有文章（Article.objects.all()）
|
|
|
articles = articles if articles else Article.objects.all()
|
|
|
# 将Django Article对象转换为Elasticsearch文档列表
|
|
|
docs = self.convert_to_doc(articles)
|
|
|
# 遍历文档列表，逐个保存到Elasticsearch
|
|
|
for doc in docs:
|
|
|
doc.save()
|
|
|
|
|
|
# 实例方法：批量更新Elasticsearch中的文章文档
|
|
|
def update_docs(self, docs):
|
|
|
# 遍历文档列表，逐个保存（已存在的文档会执行更新操作）
|
|
|
for doc in docs:
|
|
|
doc.save() |
