ContentAutomationHub

sudo apt update sudo apt install python3-pip python3-venv redis-server -y sudo systemctl start redis && sudo systemctl enable redis

 

mkdir ~/ContentAutomationHub && cd ~/ContentAutomationHub python3 -m venv venv source venv/bin/activate

pip install flask celery redis requests sqlalchemy pymysql flask-cors # 用于未来可能的API管理 pip install flask-restx

-- 任务主表 CREATE TABLE task ( id INT PRIMARY KEY AUTO_INCREMENT, title VARCHAR(255) NOT NULL COMMENT '选题标题', keywords TEXT COMMENT '关键词', status ENUM('pending', 'generating', 'review', 'publishing', 'published', 'failed') DEFAULT 'pending', ai_raw_content LONGTEXT COMMENT 'AI生成的原始内容', final_content LONGTEXT COMMENT '审核后的最终内容', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ); -- 发布平台配置表（安全存储各平台密钥） CREATE TABLE publish_platform ( id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(50) NOT NULL COMMENT '如：joomla, wechat', config JSON COMMENT '存储API URL、密钥等JSON化配置', is_active BOOLEAN DEFAULT TRUE ); -- 发布日志表 CREATE TABLE publish_log ( id INT PRIMARY KEY AUTO_INCREMENT, task_id INT, platform_id INT, status ENUM('success', 'failed'), response TEXT COMMENT '平台返回的原始响应', published_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (task_id) REFERENCES task(id), FOREIGN KEY (platform_id) REFERENCES publish_platform(id) );

# app.py from flask import Flask from celery import Celery import os # 初始化Celery celery = Celery( 'tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0' ) def create_app(): app = Flask(__name__) app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', 'your-secret-key-here') # 将Celery与App关联 celery.conf.update(app.config) # 注册蓝图（后续定义） from routes import main_bp app.register_blueprint(main_bp) return app # 创建应用实例 app = create_app()

# tasks.py from app import celery import requests import json from models import Task, db # 假设您用Flask-SQLAlchemy定义了模型 DEEPSEEK_API_URL = "https://api.deepseek.com/v1/chat/completions" DEEPSEEK_API_KEY = "your-deepseek-api-key" # 务必从环境变量读取！ @celery.task(bind=True) def generate_article(self, task_id): """任务1：调用DeepSeek生成文章""" task = Task.query.get(task_id) if not task: return task.status = 'generating' db.session.commit() headers = { 'Authorization': f'Bearer {DEEPSEEK_API_KEY}', 'Content-Type': 'application/json' } # 精心设计的提示词，是生成高质量文章的关键 prompt = f"你是一位资深的电子通信技术专家。请以'{task.title}'为题，撰写一篇深入的技术文章。要求：专业严谨、逻辑清晰、包含实例，字数在1500字左右。关键词：{task.keywords}" payload = { "model": "deepseek-chat", "messages": [{"role": "user", "content": prompt}], "temperature": 0.7 } try: response = requests.post(DEEPSEEK_API_URL, headers=headers, json=payload, timeout=60) result = response.json() article_content = result['choices'][0]['message']['content'] task.ai_raw_content = article_content task.status = 'review' # 更新状态为待审核 db.session.commit() return {"success": True, "task_id": task_id} except Exception as e: task.status = 'failed' db.session.commit() return {"success": False, "error": str(e)} @celery.task(bind=True) def publish_article(self, task_id, platform_list): """任务2：将审核后的文章发布到多个平台""" task = Task.query.get(task_id) if task.status != 'review': return task.status = 'publishing' db.session.commit() publish_results = [] for platform_name in platform_list: # 这里将调用您为每个平台编写的适配器 result = dispatch_to_platform(platform_name, task.final_content, task.title) publish_results.append(result) # 记录日志到 publish_log 表（略） task.status = 'published' db.session.commit() return publish_results def dispatch_to_platform(platform_name, content, title): # 这里是一个分发器的示例，您需要为每个平台实现具体函数 platforms = { 'joomla': publish_to_joomla, 'wechat': publish_to_wechat, # 微信公众号 # ... 其他平台 } func = platforms.get(platform_name) return func(content, title) if func else {"status": "failed", "error": "Unknown platform"}

# routes.py from flask import Blueprint, render_template, request, jsonify, redirect, url_for from models import Task, db from tasks import generate_article, publish_article main_bp = Blueprint('main', __name__) @main_bp.route('/') def index(): tasks = Task.query.order_by(Task.created_at.desc()).limit(10).all() return render_template('index.html', tasks=tasks) # 需要创建简单的前端模板 @main_bp.route('/submit_task', methods=['POST']) def submit_task(): title = request.form.get('title') keywords = request.form.get('keywords') new_task = Task(title=title, keywords=keywords, status='pending') db.session.add(new_task) db.session.commit() # 异步调用AI生成任务 generate_article.delay(new_task.id) return redirect(url_for('main.index')) @main_bp.route('/approve_and_publish/<int:task_id>', methods=['POST']) def approve_and_publish(task_id): task = Task.query.get(task_id) if task and task.status == 'review': # 假设前端通过表单传递了最终修改的内容 task.final_content = request.form.get('final_content', task.ai_raw_content) # 选择要发布的平台，例如: ['joomla', 'weibo'] platforms_to_publish = request.form.getlist('platforms') # 异步调用发布任务 publish_article.delay(task_id, platforms_to_publish) return jsonify({"status": "success", "message": "发布任务已启动"}) return jsonify({"status": "error", "message": "任务状态不允许发布"}), 400

# platforms/joomla_publisher.py import requests import json def publish_to_joomla(content, title, category_id=1): """ 发布到Joomla 假设您的Joomla已安装并配置了API插件，提供了/articles端点 """ joomla_config = { 'api_url': 'https://your-bluetooth-site.com/api/index.php/v1/articles', 'api_key': 'your-joomla-api-token' } headers = { 'X-API-KEY': joomla_config['api_key'], 'Content-Type': 'application/json' } article_data = { 'title': title, 'alias': title.replace(' ', '-').lower(), 'articletext': content, 'state': 1, # 1表示发布，0表示未发布 'catid': category_id, 'language': '*' } try: response = requests.post( joomla_config['api_url'], headers=headers, data=json.dumps(article_data), timeout=30 ) if response.status_code in [200, 201]: return {"platform": "joomla", "status": "success", "article_id": response.json().get('id')} else: return {"platform": "joomla", "status": "failed", "error": response.text} except Exception as e: return {"platform": "joomla", "status": "failed", "error": str(e)}

sudo systemctl start redis

celery -A tasks.celery worker --loglevel=info --pool=solo

          export FLASK_APP=app.py            flask run --host=0.0.0.0 --port=5000