python+openai兼容大模型api实现ai全自动垂直领域信息发布运营网站系统代码-python代码-BFW代码库

代码语言：python

所属分类：其他

代码描述：python+openai兼容大模型api实现ai全自动垂直领域信息发布运营网站系统代码，ai定时发布垂直领域信息，ai根据浏览数据自动运营网站，自动回复网友评论，可手动发布信息，支持openai兼容的大模型api。

代码标签： python openai 兼容大模型 api ai 全自动垂直领域信息发布运营网站

下面为部分代码预览，完整代码请点击下载或在bfwstudio webide中打开

# -*- coding: utf-8 -*-
"""
AI自主运营信息网站 - FastAPI 单文件实现
FastAPI + SQLite + OpenAI + 内嵌Vue3(CDN)前端 + 管理员后台
"""
import os
import json
import random
from datetime import datetime, timedelta
from functools import wraps
from collections import Counter
from typing import Optional
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request, Depends, HTTPException, Body
from fastapi.responses import HTMLResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel

from sqlalchemy import create_engine, Column, Integer, String, Text, Boolean, DateTime, ForeignKey, UniqueConstraint, func, text
from sqlalchemy.orm import declarative_base, sessionmaker, scoped_session, Session
from werkzeug.security import generate_password_hash, check_password_hash
import jwt
from apscheduler.schedulers.background import BackgroundScheduler

try:
    from openai import OpenAI
except ImportError:
    OpenAI = None

# ========================= 站点配置（按需修改） =========================
SECRET_KEY = os.environ.get("SECRET_KEY", "please-change-this-secret")
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "sk-")
OPENAI_BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1")
OPENAI_MODEL = os.environ.get("OPENAI_MODEL", "deepseek-v4-flash")

SITE_NAME = "AI前沿资讯站"
SITE_TOPIC = "人工智能、大模型与前沿科技"
CATEGORIES = ["技术突破", "行业应用", "AI大模型", "科技巨头动态", "政策与伦理"]

AUTO_PUBLISH_INTERVAL_MIN = 30
AUTO_REPLY_INTERVAL_MIN = 5

client = None
if OpenAI is not None and OPENAI_API_KEY:
    if OPENAI_BASE_URL:
        client = OpenAI(api_key=OPENAI_API_KEY, base_url=OPENAI_BASE_URL)
    else:
        client = OpenAI(api_key=OPENAI_API_KEY)

# ========================= FastAPI & DB 初始化 =========================
engine = create_engine('sqlite:///site.db', connect_args={"check_same_thread": False})
SessionLocal = scoped_session(sessionmaker(bind=engine))
Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# ========================= 数据模型 =========================
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    username = Column(String(64), unique=True, nullable=False)
    email = Column(String(120))
    password_hash = Column(String(255), nullable=False)
    is_admin = Column(Boolean, default=False) # ★ 新增管理员标识
    created_at = Column(DateTime, default=datetime.utcnow)
    last_login = Column(DateTime)

class Post(Base):
    __tablename__ = 'post'
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(Text, nullable=False)
    category = Column(String(64))
    tags = Column(String(255))
    author = Column(String(32), default="AI编辑")
    created_at = Column(DateTime, default=datetime.utcnow)
    views = Column(Integer, default=0)
    like_count = Column(Integer, default=0)

class Like(Base):
    __tablename__ = 'like'
    id = Column(Integer, primary_key=True)
    post_id = Column(Integer, ForeignKey('post.id'))
    user_id = Column(Integer, ForeignKey('user.id'))
    created_at = Column(DateTime, default=datetime.utcnow)
    __table_args__ = (UniqueConstraint('post_id', 'user_id', name='uq_like'),)

class Comment(Base):
    __tablename__ = 'comment'
    id = Column(Integer, primary_key=True)
    post_id = Column(Integer, ForeignKey('post.id'))
    user_id = Column(Integer, ForeignKey('user.id'), nullable=True)
    parent_id = Column(Integer, ForeignKey('comment.id'), nullable=True)
    content = Column(Text, nullable=False)
    is_ai = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    replied = Column(Boolean, default=False)

class BrowseHistory(Base):
    __tablename__ = 'browse_history'
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('user.id'))
    post_id = Column(Integer, ForeignKey('post.id'))
    viewed_at = Column(DateTime, default=datetime.utcnow)

class Message(Base):
    __tablename__ = 'message'
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('user.id'))
    role = Column(String(16))
    content = Column(Text)
    created_at = Column(DateTime, default=datetime.utcnow)
    replied = Column(Boolean, default=False)

# ========================= 鉴权工具 =========================
def generate_token(user_id):
    payload = {"user_id": user_id, "exp": datetime.utcnow() + timedelta(days=7)}
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

def _parse_token(request: Request) -> Optional[int]:
    auth = request.headers.get("Authorization", "")
    token = auth.replace("Bearer ", "") if auth else None
    if not token:
        return None
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload["user_id"]
    except Exception:
        return None

def get_current_user(request: Request, db: Session = Depends(get_db)):
    uid = _parse_token(request)
    if not uid:
        raise HTTPException(status_code=401, detail="未登录或token已过期")
    user = db.get(User, uid)
    if not user:
        raise HTTPException(status_code=401, detail="用户不存在")
    return user

def get_optional_user(request: Request, db: Session = Depends(get_db)):
    uid = _parse_token(request)
    if uid:
        return db.get(User, uid)
    return None

def admin_required(user: User = Depends(get_current_user)):
    if not user.is_admin:
        raise HTTPException(status_code=403, detail="需要管理员权限")
    return user

# ==========.........完整代码请登录后点击上方下载按钮下载查看