使用Django重构微信公众号爬虫与RAGFlow知识库项目

我将帮您将项目转换为Django架构，这样更适合构建完整的Web应用，并且便于后续扩展和管理。

Django项目结构规划
text
wechat-rag-django/
├── wechat_rag/ # Django项目主目录
│ ├── init.py
│ ├── settings.py # 项目设置
│ ├── urls.py # 主URL路由
│ ├── asgi.py
│ ├── wsgi.py
│ └── celery.py # Celery配置
├── crawler/ # 爬虫应用
│ ├── migrations/
│ ├── init.py
│ ├── admin.py
│ ├── apps.py
│ ├── models.py # 数据模型
│ ├── tasks.py # 异步任务
│ ├── views.py # 视图函数
│ ├── management/
│ │ └── commands/
│ │ ├── init.py
│ │ └── crawl_wechat.py # 自定义管理命令
│ └── utils.py # 工具函数
├── knowledge/ # 知识库应用
│ ├── migrations/
│ ├── init.py
│ ├── admin.py
│ ├── apps.py
│ ├── models.py
│ ├── tasks.py
│ ├── views.py
│ └── ragflow_client.py # RAGFlow客户端
├── notification/ # 通知应用
│ ├── migrations/
│ ├── init.py
│ ├── admin.py
│ ├── apps.py
│ ├── models.py
│ ├── tasks.py
│ ├── views.py
│ └── senders.py # 消息发送器
├── templates/ # 模板目录
├── static/ # 静态文件
├── requirements.txt # 依赖列表
├── manage.py # Django管理脚本
├── docker-compose.yml # Docker编排
└── Dockerfile # Docker配置
使用PyCharm创建Django项目步骤
打开PyCharm，选择"New Project"

选择"Django"项目类型

命名项目为wechat-rag-django

选择Python解释器（建议Python 3.8+）

取消勾选"Create templates directory"（我们将手动创建）

点击"Create"

项目详细配置

1. 安装必要依赖
   创建requirements.txt文件：

txt
Django>=4.2.0
celery>=5.3.0
redis>=4.5.0
requests>=2.28.0
beautifulsoup4>=4.11.0
jieba>=0.42.1
wordcloud>=1.8.1
matplotlib>=3.6.0
django-celery-beat>=2.5.0
django-celery-results>=2.5.0
psycopg2-binary>=2.9.5 # 如果使用PostgreSQL
安装依赖：

bash
pip install -r requirements.txt
2. 创建Django应用
bash
python manage.py startapp crawler
python manage.py startapp knowledge
python manage.py startapp notification
3. 配置Django设置 (wechat_rag/settings.py)
python
import os
from pathlib import Path

BASE_DIR = Path(file).resolve().parent.parent

SECRET_KEY = ‘your-secret-key-here’ # 生产环境请使用环境变量

DEBUG = True

ALLOWED_HOSTS = [‘*’]

INSTALLED_APPS = [
‘django.contrib.admin’,
‘django.contrib.auth’,
‘django.contrib.contenttypes’,
‘django.contrib.sessions’,
‘django.contrib.messages’,
‘django.contrib.staticfiles’,
‘django_celery_beat’,
‘django_celery_results’,
‘crawler’,
‘knowledge’,
‘notification’,
]

MIDDLEWARE = [
‘django.middleware.security.SecurityMiddleware’,
‘django.contrib.sessions.middleware.SessionMiddleware’,
‘django.middleware.common.CommonMiddleware’,
‘django.middleware.csrf.CsrfViewMiddleware’,
‘django.contrib.auth.middleware.AuthenticationMiddleware’,
‘django.contrib.messages.middleware.MessageMiddleware’,
‘django.middleware.clickjacking.XFrameOptionsMiddleware’,
]

ROOT_URLCONF = ‘wechat_rag.urls’

TEMPLATES = [
{
‘BACKEND’: ‘Django.template.backends.django.DjangoTemplates’,
‘DIRS’: [BASE_DIR / ‘templates’],
‘APP_DIRS’: True,
‘OPTIONS’: {
‘context_processors’: [
‘django.template.context_processors.debug’,
‘django.template.context_processors.request’,
‘django.contrib.auth.context_processors.auth’,
‘django.contrib.messages.context_processors.messages’,
],
},
},
]

WSGI_APPLICATION = ‘wechat_rag.wsgi.application’

数据库配置 - 使用SQLite开发，生产环境建议使用PostgreSQL

DATABASES = {
‘default’: {
‘ENGINE’: ‘django.db.backends.sqlite3’,
‘NAME’: BASE_DIR / ‘db.sqlite3’,
}
}

Celery配置

CELERY_BROKER_URL = ‘redis://localhost:6379/0’
CELERY_RESULT_BACKEND = ‘django-db’
CELERY_ACCEPT_CONTENT = [‘json’]
CELERY_TASK_SERIALIZER = ‘json’
CELERY_RESULT_SERIALIZER = ‘json’
CELERY_TIMEZONE = ‘Asia/Shanghai’

静态文件配置

STATIC_URL = ‘/static/’
STATICFILES_DIRS = [BASE_DIR / ‘static’]
STATIC_ROOT = BASE_DIR / ‘staticfiles’

媒体文件配置

MEDIA_URL = ‘/media/’
MEDIA_ROOT = BASE_DIR / ‘media’

国际化和本地化设置

LANGUAGE_CODE = ‘zh-hans’
TIME_ZONE = ‘Asia/Shanghai’
USE_I18N = True
USE_TZ = True

自定义配置

WECHAT_CRAWLER = {
‘PROXY’: {
‘HOST’: ‘www.16yun.cn’,
‘PORT’: ‘5445’,
‘USER’: ‘16QMSOML’,
‘PASS’: ‘280651’
},
‘HEADERS’: {
‘User-Agent’: ‘Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36’
}
}

RAGFLOW = {
‘BASE_URL’: ‘http://localhost:9380’, # RAGFlow地址
‘API_KEY’: ‘your-ragflow-api-key’ # RAGFlow API密钥
}

NOTIFICATION = {
‘EMAIL’: {
‘HOST’: ‘smtp.example.com’,
‘PORT’: 587,
‘USER’: ‘your-email@example.com’,
‘PASSWORD’: ‘your-email-password’
}
}
4. 配置Celery (wechat_rag/celery.py)
python
import os
from celery import Celery
from django.conf import settings

os.environ.setdefault(‘DJANGO_SETTINGS_MODULE’, ‘wechat_rag.settings’)

app = Celery(‘wechat_rag’)

app.config_from_object(‘django.conf:settings’, namespace=‘CELERY’)
app.autodiscover_tasks()

@app.task(bind=True)
def debug_task(self):
print(f’Request: {self.request!r}')
5. 数据模型设计 (crawler/models.py)
python
from django.db import models

class WeChatAccount(models.Model):
“”“微信公众号账号”“”
name = models.CharField(max_length=100, verbose_name=“公众号名称”)
biz = models.CharField(max_length=100, unique=True, verbose_name=“公众号BIZ”)
description = models.TextField(blank=True, verbose_name=“描述”)
is_active = models.BooleanField(default=True, verbose_name=“是否启用”)
created_at = models.DateTimeField(auto_now_add=True, verbose_name=“创建时间”)

class Meta:
    verbose_name = "微信公众号"
    verbose_name_plural = verbose_name

def __str__(self):
    return self.name

class WeChatArticle(models.Model):
“”“微信公众号文章”“”
account = models.ForeignKey(WeChatAccount, on_delete=models.CASCADE, verbose_name=“所属公众号”)
title = models.CharField(max_length=200, verbose_name=“文章标题”)
content_url = models.URLField(verbose_name=“文章链接”)
publish_time = models.DateTimeField(verbose_name=“发布时间”)
read_count = models.IntegerField(default=0, verbose_name=“阅读数”)
like_count = models.IntegerField(default=0, verbose_name=“点赞数”)
content = models.TextField(verbose_name=“文章内容”)
keywords = models.TextField(blank=True, verbose_name=“关键词”)
created_at = models.DateTimeField(auto_now_add=True, verbose_name=“创建时间”)
updated_at = models.DateTimeField(auto_now=True, verbose_name=“更新时间”)

class Meta:
    verbose_name = "微信公众号文章"
    verbose_name_plural = verbose_name
    indexes = [
        models.Index(fields=['publish_time']),
        models.Index(fields=['account', 'publish_time']),
    ]

def __str__(self):
    return self.title

6. 爬虫任务 (crawler/tasks.py)
   python
   from celery import shared_task
   from django.conf import settings
   import requests
   from bs4 import BeautifulSoup
   import jieba
   import jieba.analyse
   from .models import WeChatArticle, WeChatAccount

@shared_task
def crawl_wechat_articles(account_biz, offset=0, count=10):
“”“爬取微信公众号文章任务”“”
account = WeChatAccount.objects.get(biz=account_biz)

# 爬取文章列表
articles = get_article_list(account.biz, offset, count)

for article_data in articles:
    # 检查文章是否已存在
    if not WeChatArticle.objects.filter(content_url=article_data['content_url']).exists():
        # 获取文章详情
        content = parse_article_detail(article_data['content_url'])
        
        # 提取关键词
        keywords = extract_keywords(content)
        
        # 保存到数据库
        WeChatArticle.objects.create(
            account=account,
            title=article_data['title'],
            content_url=article_data['content_url'],
            publish_time=article_data['publish_time'],
            read_count=article_data.get('read_count', 0),
            like_count=article_data.get('like_count', 0),
            content=content,
            keywords=",".join(keywords)
        )

return f"成功爬取 {len(articles)} 篇文章"

def get_article_list(biz, offset=0, count=10):
“”“获取文章列表”“”
url = “https://mp.weixin.qq.com/mp/profile_ext”
params = {
“action”: “getmsg”,
“__biz”: biz,
“offset”: offset,
“count”: count,
}

proxy_config = settings.WECHAT_CRAWLER['PROXY']
proxies = {
    "http": f"http://{proxy_config['USER']}:{proxy_config['PASS']}@{proxy_config['HOST']}:{proxy_config['PORT']}",
    "https": f"http://{proxy_config['USER']}:{proxy_config['PASS']}@{proxy_config['HOST']}:{proxy_config['PORT']}"
}

try:
    resp = requests.get(
        url,
        params=params,
        headers=settings.WECHAT_CRAWLER['HEADERS'],
        proxies=proxies,
        timeout=10
    )
    resp.raise_for_status()
    data = resp.json()
    return data.get("general_msg_list", [])
except Exception as e:
    print(f"请求失败: {e}")
    return []

def parse_article_detail(article_url):
“”“解析文章详情”“”
try:
resp = requests.get(article_url, headers=settings.WECHAT_CRAWLER[‘HEADERS’], timeout=10)
soup = BeautifulSoup(resp.text, ‘html.parser’)
content_div = soup.find(‘div’, class_=‘rich_media_content’)
if content_div:
return content_div.get_text().strip()
return “”
except Exception as e:
print(f"解析文章详情失败: {e}")
return “”

def extract_keywords(text, topK=10):
“”“提取关键词”“”
if not text:
return []

# 加载停用词
stopwords = set()
try:
    with open("data/stopwords.txt", "r", encoding="utf-8") as f:
        for line in f:
            stopwords.add(line.strip())
except FileNotFoundError:
    pass

# 分词
words = jieba.cut(text)
# 过滤停用词和单字
filtered_words = [word for word in words if word not in stopwords and len(word) > 1]
# 提取关键词
keywords = jieba.analyse.extract_tags(" ".join(filtered_words), topK=topK)
return keywords

7. RAGFlow集成 (knowledge/ragflow_client.py)
   python
   import requests
   from django.conf import settings

class RAGFlowClient:
def init(self):
self.base_url = settings.RAGFLOW[‘BASE_URL’]
self.api_key = settings.RAGFLOW[‘API_KEY’]
self.headers = {
“Authorization”: f"Bearer {self.api_key}",
“Content-Type”: “application/json”
}

def create_knowledge_base(self, name, description=""):
    """创建知识库"""
    url = f"{self.base_url}/api/knowledge-base"
    data = {
        "name": name,
        "description": description
    }
    
    try:
        response = requests.post(url, headers=self.headers, json=data)
        response.raise_for_status()
        return response.json()
    except Exception as e:
        print(f"创建知识库失败: {e}")
        return None

def upload_document(self, kb_id, file_path, title="", tags=None):
    """上传文档到知识库"""
    url = f"{self.base_url}/api/knowledge-base/{kb_id}/document"
    
    with open(file_path, 'rb') as f:
        files = {'file': (title, f)}
        data = {
            'title': title,
            'tags': tags or []
        }
        
        try:
            response = requests.post(url, headers=self.headers, files=files, data=data)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            print(f"上传文档失败: {e}")
            return None

def query(self, kb_id, question, top_k=5):
    """查询知识库"""
    url = f"{self.base_url}/api/knowledge-base/{kb_id}/query"
    data = {
        "question": question,
        "top_k": top_k
    }
    
    try:
        response = requests.post(url, headers=self.headers, json=data)
        response.raise_for_status()
        return response.json()
    except Exception as e:
        print(f"查询知识库失败: {e}")
        return None

8. 自定义管理命令 (crawler/management/commands/crawl_wechat.py)
   python
   from django.core.management.base import BaseCommand
   from crawler.tasks import crawl_wechat_articles
   from crawler.models import WeChatAccount

class Command(BaseCommand):
help = ‘爬取微信公众号文章’

def add_arguments(self, parser):
    parser.add_argument(
        '--account',
        help='指定公众号BIZ（不指定则爬取所有启用公众号）',
    )
    parser.add_argument(
        '--offset',
        type=int,
        default=0,
        help='起始偏移量',
    )
    parser.add_argument(
        '--count',
        type=int,
        default=10,
        help='每次爬取的文章数量',
    )

def handle(self, *args, **options):
    account_biz = options['account']
    offset = options['offset']
    count = options['count']
    
    if account_biz:
        # 爬取指定公众号
        accounts = WeChatAccount.objects.filter(biz=account_biz, is_active=True)
    else:
        # 爬取所有启用公众号
        accounts = WeChatAccount.objects.filter(is_active=True)
    
    for account in accounts:
        self.stdout.write(f"开始爬取公众号: {account.name}")
        result = crawl_wechat_articles.delay(account.biz, offset, count)
        self.stdout.write(f"任务已提交: {result.id}")

9. 配置URL路由 (wechat_rag/urls.py)
   python
   from django.contrib import admin
   from django.urls import path, include

urlpatterns = [
path(‘admin/’, admin.site.urls),
path(‘crawler/’, include(‘crawler.urls’)),
path(‘knowledge/’, include(‘knowledge.urls’)),
path(‘notification/’, include(‘notification.urls’)),
]
10. 创建定时任务 (使用Django Celery Beat)
在Django Admin中配置定期任务，或使用代码配置：

python

在合适的地方（如apps.py或单独的任务配置文件中）

from django_celery_beat.models import PeriodicTask, IntervalSchedule
from celery import current_app

创建定时任务

def setup_periodic_tasks():
# 每6小时执行一次爬取任务
schedule, created = IntervalSchedule.objects.get_or_create(
every=6,
period=IntervalSchedule.HOURS,
)

PeriodicTask.objects.get_or_create(
    interval=schedule,
    name='Crawl WeChat Articles',
    task='crawler.tasks.crawl_wechat_articles',
    args='["all"]',  # 爬取所有启用公众号
)

# 每天生成一次摘要报告
daily_schedule, created = IntervalSchedule.objects.get_or_create(
    every=1,
    period=IntervalSchedule.DAYS,
)

PeriodicTask.objects.get_or_create(
    interval=daily_schedule,
    name='Generate Daily Digest',
    task='notification.tasks.generate_daily_digest',
)

部署到云服务器
Docker配置
创建Dockerfile:

dockerfile
FROM python:3.10-slim

ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1

WORKDIR /app

RUN apt-get update && apt-get install -y
gcc
libpq-dev
&& rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

RUN python manage.py collectstatic --noinput

EXPOSE 8000

CMD [“gunicorn”, “–bind”, “0.0.0.0:8000”, “wechat_rag.wsgi:application”]
创建docker-compose.yml:

yaml
version: ‘3.8’

services:
web:
build: .
command: gunicorn --bind 0.0.0.0:8000 wechat_rag.wsgi:application
volumes:
- .:/app
- static_volume:/app/staticfiles
- media_volume:/app/media
ports:
- “8000:8000”
env_file:
- .env
depends_on:
- redis
- db

db:
image: postgres:13
volumes:
- postgres_data:/var/lib/postgresql/data/
environment:
- POSTGRES_DB=wechat_rag
- POSTGRES_USER=wechat_rag_user
- POSTGRES_PASSWORD=wechat_rag_password

redis:
image: redis:7-alpine

celery:
build: .
command: celery -A wechat_rag worker --loglevel=info
volumes:
- .:/app
env_file:
- .env
depends_on:
- redis
- db

celery-beat:
build: .
command: celery -A wechat_rag beat --loglevel=info
volumes:
- .:/app
env_file:
- .env
depends_on:
- redis
- db

volumes:
postgres_data:
static_volume:
media_volume:
部署脚本
创建deploy.sh:

bash
#!/bin/bash

设置环境变量

export SECRET_KEY=“your-django-secret-key”
export DEBUG=“False”
export DATABASE_URL=“postgres://wechat_rag_user:wechat_rag_password@db:5432/wechat_rag”
export RAGFLOW_BASE_URL=“http://your-ragflow-instance:9380”
export RAGFLOW_API_KEY=“your-ragflow-api-key”

构建并启动服务

docker-compose up -d --build

执行数据库迁移

docker-compose exec web python manage.py migrate

创建超级用户

docker-compose exec web python manage.py createsuperuser --noinput --username admin --email admin@example.com || true

收集静态文件

docker-compose exec web python manage.py collectstatic --noinput

echo “部署完成!”
使用说明
初始化项目:

bash

创建超级用户

python manage.py createsuperuser

执行数据库迁移

python manage.py migrate

启动开发服务器

python manage.py runserver
添加微信公众号:

访问Django Admin (/admin)

添加微信公众号记录（名称和BIZ参数）

执行爬取任务:

bash

爬取所有公众号

python manage.py crawl_wechat

爬取指定公众号

python manage.py crawl_wechat --account=公众号BIZ
配置定时任务:

访问Django Admin中的Periodic Tasks

配置爬取频率和其他定时任务

访问API:

文章列表: /crawler/articles/

知识库查询: /knowledge/query/

这个Django项目架构提供了完整的Web管理界面、API接口和定时任务功能，非常适合部署到云服务器并与其他系统（如RAGFlow）集成。

代码

https://github.com/1195010543/wechat-rag-django

云服务器地址
https://59.110.45.9:8888/database
使用宝塔面板

问题：
1、截取的数据不全
2、知识库读DB的数据不确定可不可以，要不还是先放到excel文件里面
3、rag部署到云端的docker看下可以不，需要在服务器上面安装docker