【云原生| Docker】 部署 Django & mysql 项目

举报
计算机魔术师 发表于 2022/08/10 10:07:06 2022/08/10
【摘要】 @[toc] 一、爬取帖子、二级评论爬取源码from lxml import etree # 导入所需模块import requests, reroot_url = "https://tieba.baidu.com"headers = { 'Cookie': '你的cookie', 'Accept': 'image/avif,image/webp,image/apng,imag...

@[toc]

一、爬取帖子、二级评论

爬取源码

from lxml import etree  # 导入所需模块
import requests, re

root_url = "https://tieba.baidu.com"

headers = {
    'Cookie': '你的cookie',
    'Accept': 'image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=0.8',
    'Accept-Encoding': 'gzip, deflate, br',
    'Accept-Language': 'zh-CN,zh;q=0.9'
}

# 帖吧(第一页)
base_comment = []
page_first = 0  # 第一页
while page_first < 5:  # 默认爬取五页

    resp = requests.get('https://tieba.baidu.com/f?kw=佛山科学技术学院&ie=utf-8' + f'&pn={page_first * 50}', headers=headers)
    html_baidu = resp.content.decode('utf-8')  # 手动解码  text默认返回猜测的解码方式
    # 根路由
    tree_baidu = etree.HTML(html_baidu)
    divs_daidu = tree_baidu.xpath('//*[@id="thread_list"]/li')

    # 获取帖子字典
    for div in divs_daidu:
        dict_temp = {}
        if div.xpath('./div/div[2]/div[1]/div[1]/a/@href'):
            dict_temp['id'] = re.sub(r'/p/', '', div.xpath('./div/div[2]/div[1]/div[1]/a/@href')[0])
            dict_temp['comment'] = div.xpath('./div/div[2]/div[1]/div[1]/a/text()')
            dict_temp['href'] = root_url + div.xpath('./div/div[2]/div[1]/div[1]/a/@href')[0]
            dict_temp['img'] = div.xpath(
                f'//*[@id="fm{re.sub(r"/p/", "", div.xpath("./div/div[2]/div[1]/div[1]/a/@href")[0])}"]/li/a/img/@data-original')
            base_comment.append(dict_temp)

    # 二级评论参数
    second_page = 1
    param_second = {
        'pn': str(second_page)
    }
    first_pagesize = 0
    # todo 二轮循环 帖子收集
    while first_pagesize < len(base_comment):
        second_pagesize = 0
        # todo  一轮循环 -> 帖子回复
        resp_second = requests.get(url=base_comment[first_pagesize]['href'], params=param_second)  # 获取响应
        second_html = resp_second.content.decode('utf-8')  # 获取源码
        second_tree = etree.HTML(second_html)  # 解析源代码
        second_root = second_tree.xpath('//*[@id="j_p_postlist"]/div')  # 获取节点树

        while second_pagesize < len(second_root):
            second_comments = []  # 收集所有二级评论
            second_comment_dict = {}
            for comment in second_root:
                if comment.xpath('./div[2]/ul/li[3]/a/@href'):  # 空字符串是广告
                    # print(comment.xpath('./div[2]/ul/li[3]/a/text()'))
                    # print(root_url + comment.xpath('./div[2]/ul/li[3]/a/@href')[0])
                    print(comment.xpath('./div[3]/div[1]/cc/div[2]')[0].xpath("string(.)").strip())

                    second_comment_dict = {
                        'user': comment.xpath('./div[2]/ul/li[3]/a/text()'),  # 用户名
                        'user_url': root_url + comment.xpath('./div[2]/ul/li[3]/a/@href')[0],  # 用户主页
                        'comment': comment.xpath('./div[3]/div[1]/cc/div[2]')[0].xpath("string(.)").strip()
                    }  # 评论内容
                if second_comment_dict not in second_comments:  # 去重
                    second_comments.append(second_comment_dict)
            base_comment[first_pagesize]['second_level'] = second_comments

            second_pagesize += 1  # 二级评论数
            print('下一条评论')
        first_pagesize += 1  # 贴子条数
        print('下一条帖子')
    page_first += 1  # 贴吧页数
    print('下一页帖吧')
print('over!')

爬取数据结果：

在这里插入图片描述
注意： 爬取只做舆情监控，不做其他用途，不要用于恶意爬取（爬取次数多，百度也会出现安全验证）

二、构建数据表

创建应用后，我们根据爬取数据格式建立表格：

{'id': '7946129745',  //贴吧id
  'comment': ['当你们看见这个帖子的时候'], //标题
  'href': 'https://tieba.baidu.com/p/7946129745', // 贴吧链接
  'img':    //图片
   ['https://tiebapic.baidu.com/forum/wh%3D200%2C90%3B/sign=d20c17c12559252da342150604ab2f02/377adab44aed2e73802173f7c201a18b86d6fa25.jpg?tbpicau=2022-08-09-05_20b9d8c790e4c21592a066f388f2804a'],
  'second_level': 
  [{'user': ['我是校长-🤗'],  // 评论用户
    'user_url': 'https://tieba.baidu.com/home/main?id=tb.1.e2e8d390.m1upwH4nNfqXXxitUEGaqg&fr=pb&ie=utf-8',  // 用户信息地址
    'comment': '我就11级啦'},   // 品论内容
   {'user': ['我是校长-🤗'],  // 多份评论
     ····
    }，
    ····
    ]
 }

建立如下表以存贮舆情数据：

  1. 贴吧用户 ( 从评论中获取）
    a. 用户名
    b. 贴吧用户url

代码：

class Baidu_User(models.Model):
    """
    a. 用户名
    b. 贴吧用户`url`
    """
    username = models.CharField(max_length=128, verbose_name='贴吧用户名', blank=True, null=True)
    url = models.CharField(max_length=200, verbose_name='贴吧用户信息', blank=True, null=True)
    create_time = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')
    modify_time = models.DateTimeField(auto_now=True, verbose_name='修改时间')

    def __str__(self):
        return self.username

    class Meta:
        db_table = "baidu_user"
        verbose_name = _('贴吧用户')
        verbose_name_plural = _('贴吧用户')  # 复数形式
  1. 帖子 （ 从最外层获取）
    a. 帖子唯一id
    b. 帖子标题
    c. 帖子照片
    d. 帖子路由url
class Baidu_Post(models.Model):
    """
    a. 帖子唯一`id`
    b. 帖子标题
    c. 帖子照片路由
    d. 帖子路由`url`
    e. 一对多外键 → 评论
    """
    post_id = models.CharField(max_length=128, verbose_name='帖子ID', blank=True)
    title = models.CharField(max_length=1024, verbose_name='标题', blank=True)
    img_url = models.CharField(max_length=2048, verbose_name='图片', blank=True)
    url = models.CharField(max_length=200, verbose_name='帖子地址', blank=True)
    create_time = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')
    modify_time = models.DateTimeField(auto_now=True, verbose_name='修改时间')

    def __str__(self):
            return self.title

    class Meta:
        db_table = "baidu_post"
        verbose_name = _('贴子')
        verbose_name_plural = _('帖子')  # 复数形式
  1. 评论
    a. 一对一外键 → 贴吧用户
    b. 内容
    c. 多对一外键 → 帖子
class Baidu_Comment(models.Model):
    """
    a. 一对一外键 → 贴吧用户
    b. 内容
    c. 多对一外键 → 帖子
    """
    baidu_user = models.OneToOneField(Baidu_User, on_delete=models.CASCADE, related_name="comment", verbose_name='百度用户')
    comment = models.CharField(max_length=2048, verbose_name='评论', blank=True)
    baidu_post = models.ForeignKey(Baidu_Post, on_delete=models.CASCADE, related_name="comment", verbose_name='对应帖子')
    create_time = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')
    modify_time = models.DateTimeField(auto_now=True, verbose_name='修改时间')

    def __str__(self):
        return self.comment

    class Meta:
        db_table = "baidu_comment"
        verbose_name = _('评论')
        verbose_name_plural = _('评论')  # 复数形式

数据迁移并在后台注册

三、并入项目

1. view视图代码

from django.shortcuts import render

# Create your views here.

from lxml import etree  # 导入所需模块
import requests, re
from .models import Baidu_User, Baidu_Post, Baidu_Comment
from django.contrib import messages
from django.shortcuts import HttpResponseRedirect
root_url = "https://tieba.baidu.com"

headers = {
    'Cookie': '你的cookie',
    'Accept': 'image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=0.8',
    'Accept-Encoding': 'gzip, deflate, br',
    'Accept-Language': 'zh-CN,zh;q=0.9'
}

# 二级评论参数
second_page = 1  # 默认只爬取第一页
param_second = {
    'pn': str(second_page)
}


def collect_baidu(request):
    # 获取帖吧内容
    # base_comment = []
    page_first = 0  # 第一页
    while page_first < 5:  # 默认爬取五页
        resp = requests.get('https://tieba.baidu.com/f?kw=佛山科学技术学院&ie=utf-8' + f'&pn={page_first * 50}',
                            headers=headers)
        html_baidu = resp.content.decode('utf-8')  # 手动解码  text默认返回猜测的解码方式
        # 根路由
        tree_baidu = etree.HTML(html_baidu)
        divs_daidu = tree_baidu.xpath('//*[@id="thread_list"]/li')

        # 获取帖子字典
        for div in divs_daidu:
            dict_temp = {}
            if bool(div.xpath('./div/div[2]/div[1]/div[1]/a/@href')):
                dict_temp['id'] = re.sub(r'/p/', '', div.xpath('./div/div[2]/div[1]/div[1]/a/@href')[0])
                dict_temp['comment'] = div.xpath('./div/div[2]/div[1]/div[1]/a/text()')
                dict_temp['href'] = root_url + div.xpath('./div/div[2]/div[1]/div[1]/a/@href')[0]
                dict_temp['img'] = div.xpath(
                    f'//*[@id="fm{re.sub(r"/p/", "", div.xpath("./div/div[2]/div[1]/div[1]/a/@href")[0])}"]/li/a/img/@data-original')
                # base_comment.append(dict_temp)
                #  保存到数据库
                if not Baidu_Post.objects.filter(post_id=dict_temp['id']).first():
                    baidu_post = Baidu_Post.objects.create(
                        post_id=dict_temp['id'],
                        title=dict_temp['comment'][0],
                        url=dict_temp['href'],
                        img_url=dict_temp['img']
                    )
                    baidu_post.save()

                resp_second = requests.get(url=dict_temp['href'], params=param_second)  # 获取响应
                second_html = resp_second.content.decode('utf-8')  # 获取源码
                second_tree = etree.HTML(second_html)  # 解析源代码
                second_root = second_tree.xpath('//*[@id="j_p_postlist"]/div')  # 获取节点树

                second_comments = []  # 收集所有二级评论
                second_comment_dict = {}
                for comment in second_root:
                    if comment.xpath('./div[2]/ul/li[3]/a/@href'):  # 空字符串是广告

                        second_comment_dict = {
                            'user': comment.xpath('./div[2]/ul/li[3]/a/text()'),  # 用户名
                            'user_url': root_url + comment.xpath('./div[2]/ul/li[3]/a/@href')[0],  # 用户主页
                            'comment': comment.xpath('./div[3]/div[1]/cc/div[2]')[0].xpath("string(.)").strip()
                        }  # 评论内容
                    if second_comment_dict not in second_comments:  # 去重
                        second_comments.append(second_comment_dict)
                        #  保存到数据库
                        if not bool(Baidu_User.objects.filter(url=second_comment_dict['user_url']).first()):
                            baidu_user = Baidu_User.objects.create(
                                username=second_comment_dict['user'][0],
                                url=second_comment_dict['user_url']
                            )
                            baidu_user.save()
                            baidu_comment = Baidu_Comment.objects.create(
                                baidu_user=baidu_user,
                                comment=second_comment_dict['comment'],
                                baidu_post=baidu_post
                            )
                            baidu_comment.save()

                print('帖子id: {},标题：{},帖子地址:{},图片:{},评论：{}'.format(dict_temp['id'],
                                                                  dict_temp['comment'],
                                                                  dict_temp['href'],
                                                                  dict_temp['img'],
                                                                  second_comments))
        page_first += 1  # 贴吧页数

    print('over!')
    messages.add_message(request, '刷新数据成功', level=messages.SUCCESS,)
    return HttpResponseRedirect('/admin')

效果：
💯:[图片]

💯:[图片]

2. 优化后台界面

admin 代码

from django.contrib import admin
from .models import Baidu_User, Baidu_Post, Baidu_Comment
from django.utils.safestring import mark_safe
import re


# Register your models here.


@admin.register(Baidu_User)
class Baidu_UserAdmin(admin.ModelAdmin):
    # 展示列表
    list_display = ('username', 'url_tag', 'create_time')
    list_display_links = ('username',)
    readonly_fields = ['username', 'url', ]
    date_hierarchy = 'create_time'  # 根据创建时间划分等级
    ordering = ['create_time', ]  # 默认按照最新时间排序

    def url_tag(self, obj):
        if obj.url:
            return mark_safe(  # obj.picture 是相对路径, obj.picture.url是完整路径
                f'<a href="{obj.url}" target="_blank" >用户贴吧地址</a>')
        return '-'

    url_tag.short_description = '贴吧用户地址'


@admin.register(Baidu_Post)
class Baidu_PostAdmin(admin.ModelAdmin):
    # 展示列表
    list_display = ('title', 'img_tag', 'url_tag', 'create_time')
    # list_display_links = ('title',)
    exclude = ('img_url',)
    search_fields = ('title',)  # 设置可搜索内容
    search_help_text = '搜索帖子'  # 搜索提示文本， 默认为False
    readonly_fields = ['post_id', 'title', 'url', ]
    date_hierarchy = 'create_time'  # 根据创建时间划分等级
    ordering = ['create_time', ]  # 默认按照最新时间排序

    def img_tag(self, obj):
        img_html = ''
        if obj.img_url != '[]':
            temp = re.sub("'", '520', obj.img_url)
            img_list = re.findall('520(.*?)520', temp)  # 在字符串提取内容
            for img in img_list:
                img_html += f'<image src="{img}" style="width:80px; height:80px;" alt="图片" />'
            return mark_safe(img_html)
        return '-'

    img_tag.short_description = '帖子图片'

    def url_tag(self,obj):
        if obj.url:
            return mark_safe(f"<a href='{obj.url}' target='_blank'>{obj.url}</a>")
        return '-'
    url_tag.short_description = "帖子地址"

@admin.register(Baidu_Comment)
class Baidu_CommentAdmin(admin.ModelAdmin):
    # 展示列表
    list_display = ('comment', 'baidu_post', 'baidu_user', 'create_time')
    # list_display_links = ('comment',)
    search_fields = ('comment', 'baidu_user__username', 'baidu_post__title')  # 设置搜索栏范围，如果有外键，要注明外键的哪个字段，双下划线
    search_help_text = '搜索帖子评论或者用户评论记录'  # 搜索提示文本， 默认为False
    readonly_fields = ['baidu_user', 'comment', 'baidu_post', ]
    date_hierarchy = 'create_time'  # 根据创建时间划分等级
    ordering = ['create_time', ]  # 默认按照最新时间排序

效果：
在这里插入图片描述

在这里插入图片描述

在这里插入图片描述

3. urls路由

from django.contrib import admin
from django.urls import re_path
from . import views

app_name = 'public_opinion'

urlpatterns = [
    re_path(r'^$', views.collect_baidu, name="crawler")
]

测试效果
在这里插入图片描述

四、定时任务爬取

定时爬取贴吧评论进行舆情监控

4.1 安装

官方的 github地址 https://github.com/jcass77/django-apscheduler
官方文档https://apscheduler.readthedocs.io/en/latest/userguide.html#choosing-the-right-scheduler-job-store-s-executor-s-and-trigger-s

pip install django-apscheduler`

4.2 配置

在 setting.py 的 app 里面加入

INSTALLED_APPS = (
    # ...
    "django_apscheduler",
)

apscheduler 存在数据库依赖. migrate 一下
会在数据库中生成两行表

在这里插入图片描述

表结构在下面的使用中再说
在这里插入图片描述
在这里插入图片描述

4.3 简单的原理解释

注册后的任务会根据 id 以及相关的配置进行定时任务, 定时任务分为两种

  1. 定时循环任务 (循环不断触发 比如每 20s 执行一次)

  2. 定点执行任务 (只触发一次 比如 周一执行一次)

  3. 定时定点循环任务 (比如 每周一执行一次)
    任务的执行存在撞车的可能性, 即上一次任务没执行完, 下次任务就来了
    此时下次任务会被错过 (missed) 上次任务不执行完. 下次任务永远都会错过

4.4 书写代码

可以在任意view.py中实现代码 ( 通常习惯将所有定时任务放在一个APP里）

这里设置为每天早上8：30定时爬取评论！

from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore, register_job


# 实例化调度器(后台运行，不阻塞）
scheduler = BackgroundScheduler()

try:
    # 设置定时任务
    @register_job(scheduler, 'cron', id='baidu', hour=8, minute=30, day='Monday')
    def collect_baidu(request):
       pass
    # 5.开启定时任务
    scheduler.start()

except Exception as e:
    print(e)
    # 有错误停止运行
    scheduler.shutdown()

接着注释掉用于测试的路由url

4.5 执行结果查看

在admin 里面查看也较为方便, 对 admin 进行了相应的封装方便过滤啥的, 以及状态清晰
在这里插入图片描述

django_apscheduler_djangojob 表保存注册的任务以及下次执行的时间
在这里插入图片描述

django_apscheduler_djangojobexecution 保存每次任务执行的时间和结果和任务状态
在这里插入图片描述

这里注意 missed 则是表示撞车的场景, 为避免这种场景需要在 周期的长度以及是否进行强制结束进行选择

4.6 其他问题

APScheduler 在 uwsgi 环境下会有存在复数个进程同时启用的问题， 从而重复执行占用资源

解决方式
https://www.cnblogs.com/shijieli/p/15528080.html

博客园定时任务参考文章
csdn定时任务讲解文章
源码讲解文章
腾讯云定时任务详解

【版权声明】本文为华为云社区用户原创内容，转载时必须标注文章的来源（华为云社区）、文章链接、文章作者等基本信息， 否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱： cloudbbs@huaweicloud.com
Django MySQL 云原生
  • 点赞
  • 收藏
  • 关注作者

评论（0）

0/1000
抱歉，系统识别当前为高风险访问，暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称，即可参与社区互动！

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

确认 取消

加入云驻计划，成为创作者

  • 华为云周边好礼
  • 免费体验产品
  • 特殊身份标识
  • 线下官方门票
  • 内部专家零距离
  • 与10000+优质创作者共同成长
立即加入