源码下载地址：https://download.csdn.net/download/mosquito_lover1/90285144
系统截图：
功能介绍：
支持本地化部署，支持上传pdf、word、txt等格式文件，支持对文件名和文件内容的检索。
核心源码：
from flask import Blueprint, render_template, redirect, url_for, flash, request, abort, send_file, jsonify
from flask_login import login_user, logout_user, login_required, current_user
from app.models import User, Article, Document, Question, QALog
from app import db
import os
from datetime import datetime, timedelta
from werkzeug.utils import secure_filename
from config import Config
from docx import Document as DocxDocument
from PyPDF2 import PdfReader
from pptx import Presentation
from openpyxl import load_workbook
import io
from sqlalchemy import or_, func
import requests
from bs4 import BeautifulSoup
import re
main = Blueprint('main', __name__)
def get_hot_questions(limit=3):
"""获取最近一周的热门问题"""
one_week_ago = datetime.now() - timedelta(days=7)
# 统计最近一周内的热门问题
hot_questions = db.session.query(
QALog.question,
func.count(QALog.question).label('count')
).filter(
QALog.created_at >= one_week_ago,
QALog.source != '无匹配' # 排除未找到答案的问题
).group_by(
QALog.question
).order_by(
func.count(QALog.question).desc()
).limit(limit).all()
return hot_questions
def get_real_ip():
"""获取用户真实IP地址"""
# 按优先级尝试不同的请求头
headers_to_check = [
'X-Forwarded-For',
'X-Real-IP',
'CF-Connecting-IP', # Cloudflare
'True-Client-IP'
]
for header in headers_to_check:
ip = request.headers.get(header)
if ip:
# X-Forwarded-For 可能包含多个 IP，取第一个
return ip.split(',')[0].strip()
# 如果没有代理头信息，则使用远程地址
return request.remote_addr or '127.0.0.1'
@main.route('/', methods=['GET', 'POST'])
def index():
answer = None
query = None
source = None
source_id = None
if request.method == 'POST':
query = request.form.get('query', '').strip()
if query:
# 1. 首先尝试基础问答
basic_answer = get_basic_answer(query)
if basic_answer:
answer = basic_answer
source = "基础问答"
else:
# 2. 搜索预设问答
questions = Question.query.all()
for q in questions:
if query.lower() in q.question.lower():
answer = q.answer
source = "预设问答"
source_id = q.id
break
if q.keywords and any(k.strip().lower() in query.lower() for k in q.keywords.split(',')):
answer = q.answer
source = "预设问答"
source_id = q.id
break
# 3. 如果没找到，搜索文章内容
if not answer:
articles = Article.query.all()
for article in articles:
if query.lower() in article.content.lower():
paragraphs = article.content.split('\n')
for para in paragraphs:
if para.strip() and query.lower() in para.lower():
answer = para.strip()
source = "文章"
source_id = article.id
break
if answer:
break
# 4. 如果还没找到，搜索文档内容
if not answer:
documents = Document.query.all()
for doc in documents:
file_path = os.path.join(Config.UPLOAD_FOLDER, doc.filename)
if os.path.exists(file_path):
content = extract_text_from_file(file_path, doc.file_type.lower())
if content and query.lower() in content.lower():
paragraphs = content.split('\n')
for para in paragraphs:
if para.strip() and query.lower() in para.lower():
max_length = 500
if len(para) > max_length:
start = para.lower().find(query.lower())
start = max(0, start - 100)
end = min(len(para), start + max_length)
para = '...' + para[start:end] + '...'
answer = para.strip()
source = "文档"
source_id = doc.id
break
if answer:
break
if not answer:
answer = "抱歉，我没有找到相关的答案。"
source = "无匹配"
# 记录问答日志
qa_log = QALog(
question=query,
answer=answer,
source=source,
source_id=source_id,
ip_address=get_real_ip(),
created_at=datetime.now() # 使用当前时间
)
db.session.add(qa_log)
db.session.commit()
# 获取热门问题
hot_questions = get_hot_questions()
# 只有管理员才能看到文章列表
articles = None
if current_user.is_authenticated:
articles = Article.query.order_by(Article.created_at.desc()).all()
return render_template('index.html',
articles=articles,
answer=answer,
query=query,
hot_questions=hot_questions)
@main.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'POST':
username = request.form.get('username')
password = request.form.get('password')
if not username or not password:
flash('请输入用户名和密码')
return redirect(url_for('main.login'))
user = User.query.filter_by(username=username).first()
if user is None:
flash('用户名不存在')
return redirect(url_for('main.login'))
if not user.check_password(password):
flash('密码错误')
retur
标签：11,lower,para,AI,source,源码,import,answer,query From： https://blog.csdn.net/mosquito_lover1/article/details/145235321