commit b7b09251f8cf30b41e1797127e5302cd25d95cf8 Author: hz4th_coder Date: Sun Jul 12 01:07:26 2026 +0800 初始化参数数据自动化管理系统 功能: - 文章内容库管理 - 待处理产品列表管理 - 自动处理流程 - 智能搜索和数据提取 - ParamHub API集成 - 定时任务调度 部署端口: 16043 diff --git a/README.md b/README.md new file mode 100644 index 0000000..2af7550 --- /dev/null +++ b/README.md @@ -0,0 +1,270 @@ +# Param Auto Manager - 参数数据自动化管理系统 + +自动管理参数数据网站的系统，支持从内容库和互联网搜索数据，自动处理产品并提交到后台管理待审核区。 + +## 功能特性 + +- 📚 **内容库管理**：存储和管理搜索获取的相关文章内容 +- 🔄 **自动处理流程**：自动从待处理列表中提取产品并处理 +- 🔍 **智能搜索**：从内容库和互联网搜索相关数据 +- 📝 **数据提取**：根据产品类别提取和填充字段 +- ✅ **审核提交**：自动提交到ParamHub后台管理待审核区 +- 🕐 **定时任务**：支持定时自动处理产品 + +## 系统架构 + +``` +param-auto-manager/ +├── app.py # 主应用入口 +├── config.py # 配置文件 +├── requirements.txt # Python依赖 +├── models/ +│ └── database.py # 数据库模型 +├── routes/ +│ ├── articles.py # 文章内容库API +│ ├── products.py # 产品处理API +│ └── system.py # 系统管理API +├── services/ +│ ├── search_service.py # 搜索服务 +│ ├── process_service.py # 数据处理服务 +│ └── paramhub_client.py # ParamHub API客户端 +├── utils/ +│ └── scheduler.py # 定时任务调度器 +├── data/ # 数据库文件目录 +└── logs/ # 日志文件目录 +``` + +## 安装依赖 + +```bash +cd works/param-auto-manager +pip install -r requirements.txt +``` + +## 启动服务 + +```bash +# 方式1: 使用启动脚本 +chmod +x start.sh +./start.sh + +# 方式2: 直接运行 +python3 app.py +``` + +服务将在端口 **16043** 启动。 + +## API 文档 + +### 文章内容库 API (`/api/articles`) + +#### 获取文章列表 +``` +GET /api/articles?limit=100&offset=0 +``` + +#### 搜索文章 +``` +GET /api/articles/search?q=关键词&category=分类 +``` + +#### 获取文章详情 +``` +GET /api/articles/{article_id} +``` + +#### 创建文章 +``` +POST /api/articles +Content-Type: application/json + +{ + "product_names": ["产品A", "产品B"], + "category": "AI模型", + "keywords": ["关键词1", "关键词2"], + "summary": "文章摘要", + "content": "文章内容", + "source": "来源", + "url": "原文链接" +} +``` + +#### 从URL抓取文章 +``` +POST /api/articles/fetch +Content-Type: application/json + +{ + "url": "https://example.com/article", + "product_names": ["产品A"], + "category": "分类" +} +``` + +#### 删除文章 +``` +DELETE /api/articles/{article_id} +``` + +### 产品处理 API (`/api/products`) + +#### 获取待处理产品列表 +``` +GET /api/products/pending?limit=20&order_by=priority +``` + +#### 添加待处理产品 +``` +POST /api/products/pending +Content-Type: application/json + +{ + "product_name": "产品名称", + "category": "分类", + "subcategory": "子分类", + "priority": 10, + "source": "manual" +} +``` + +#### 批量添加待处理产品 +``` +POST /api/products/pending +Content-Type: application/json + +[ + {"product_name": "产品1", "category": "分类"}, + {"product_name": "产品2", "category": "分类"} +] +``` + +#### 移除待处理产品 +``` +DELETE /api/products/pending/{product_name} +``` + +#### 获取处理中的产品 +``` +GET /api/products/processing +``` + +#### 获取处理历史 +``` +GET /api/products/history?limit=100 +``` + +#### 获取指定产品的处理历史 +``` +GET /api/products/history/{product_name} +``` + +#### 处理单个产品 +``` +POST /api/products/process +Content-Type: application/json + +{ + "product_name": "产品名称", + "category": "分类", + "subcategory": "子分类" +} +``` + +#### 批量处理产品 +``` +POST /api/products/process/batch +Content-Type: application/json + +{ + "limit": 5 +} +``` + +### 系统管理 API (`/api/system`) + +#### 获取系统配置 +``` +GET /api/system/config +``` + +#### 更新系统配置 +``` +PUT /api/system/config +Content-Type: application/json + +{ + "auto_process_enabled": "true", + "process_interval": "300", + "batch_size": "5" +} +``` + +#### 获取系统统计 +``` +GET /api/system/stats +``` + +#### 健康检查 +``` +GET /api/system/health +``` + +## 处理流程 + +1. **添加待处理产品**：手动添加或系统自动发现新产品 +2. **自动/手动触发处理**： + - 从内容库搜索相关文章 + - 从互联网搜索最新数据 + - 提取产品具体内容 + - 根据类别字段填充数据 + - 提交到ParamHub待审核区 +3. **发现新产品**：处理过程中自动发现并添加相关产品 + +## 数据库表结构 + +### articles (文章内容库) +- id, product_names, category, keywords, summary, content, source, url, fetch_date + +### pending_products (待处理产品列表) +- id, product_name, category, subcategory, priority, source + +### processing_products (处理中产品) +- id, product_name, category, subcategory, status, started_at + +### process_history (处理历史) +- id, product_name, category, subcategory, status, review_id, details + +## 配置说明 + +编辑 `config.py` 文件： + +```python +PORT = 16043 # 服务端口 +PARAMHUB_BASE_URL = 'http://localhost:16041' # ParamHub API地址 +PARAMHUB_PASSWORD = 'admin123' # ParamHub管理员密码 +SEARCH_MAX_RESULTS = 10 # 搜索最大结果数 +PROCESS_INTERVAL = 300 # 自动处理间隔(秒) +BATCH_SIZE = 5 # 批量处理数量 +``` + +## 注意事项 + +1. 确保 ParamHub 服务（端口16041）正常运行 +2. 首次运行会自动创建数据库和表结构 +3. 定时任务默认每5分钟执行一次自动处理 +4. 可通过系统配置API调整自动处理参数 + +## 日志 + +日志文件位于 `logs/app.log`，包含： +- 系统启动信息 +- 处理过程记录 +- 错误和异常信息 + +## 版本历史 + +- v1.0.0 (2026-07-12): 初始版本 + - 内容库管理 + - 产品处理流程 + - 定时任务调度 + - ParamHub API集成 \ No newline at end of file diff --git a/__pycache__/config.cpython-310.pyc b/__pycache__/config.cpython-310.pyc new file mode 100644 index 0000000..b9866f6 Binary files /dev/null and b/__pycache__/config.cpython-310.pyc differ diff --git a/app.pid b/app.pid new file mode 100644 index 0000000..02fb4cc --- /dev/null +++ b/app.pid @@ -0,0 +1 @@ +2706006 diff --git a/app.py b/app.py new file mode 100644 index 0000000..831fc8b --- /dev/null +++ b/app.py @@ -0,0 +1,80 @@ +""" +参数数据自动化管理系统 +""" +from flask import Flask, jsonify +from flask_cors import CORS +from config import Config +from models.database import db +from utils.scheduler import task_scheduler, setup_auto_process_job +import logging +import os + +# 配置日志 +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + handlers=[ + logging.FileHandler(os.path.join(Config.LOG_DIR, 'app.log')), + logging.StreamHandler() + ] +) +logger = logging.getLogger('param_auto_manager') + +# 创建Flask应用 +app = Flask(__name__) +CORS(app) + +# 注册路由 +from routes.articles import bp as articles_bp +from routes.products import bp as products_bp +from routes.system import bp as system_bp + +app.register_blueprint(articles_bp) +app.register_blueprint(products_bp) +app.register_blueprint(system_bp) + +# 首页 +@app.route('/') +def index(): + return jsonify({ + 'name': 'Param Auto Manager', + 'version': '1.0.0', + 'description': '参数数据自动化管理系统', + 'endpoints': { + 'articles': '/api/articles', + 'products': '/api/products', + 'system': '/api/system' + } + }) + +# 错误处理 +@app.errorhandler(404) +def not_found(error): + return jsonify({'error': '资源不存在'}), 404 + +@app.errorhandler(500) +def internal_error(error): + return jsonify({'error': '服务器内部错误'}), 500 + +if __name__ == '__main__': + # 确保日志目录存在 + os.makedirs(Config.LOG_DIR, exist_ok=True) + os.makedirs(os.path.dirname(Config.DATABASE), exist_ok=True) + + # 初始化数据库 + logger.info("初始化数据库...") + + # 设置自动处理任务 + setup_auto_process_job() + + # 启动定时任务调度器 + task_scheduler.start() + logger.info("定时任务调度器已启动") + + # 启动Flask应用 + logger.info(f"启动服务，端口: {Config.PORT}") + app.run( + host=Config.HOST, + port=Config.PORT, + debug=Config.DEBUG + ) \ No newline at end of file diff --git a/config.py b/config.py new file mode 100644 index 0000000..f004df0 --- /dev/null +++ b/config.py @@ -0,0 +1,28 @@ +""" +配置文件 +""" +import os + +class Config: + # 应用配置 + PORT = 16043 + HOST = '0.0.0.0' + DEBUG = True + + # 数据库配置 + DATABASE = os.path.join(os.path.dirname(__file__), 'data', 'param_auto.db') + + # ParamHub API 配置 + PARAMHUB_BASE_URL = 'http://localhost:16041' + PARAMHUB_PASSWORD = 'admin123' + + # 搜索配置 + SEARCH_MAX_RESULTS = 10 # 每次搜索最多返回结果数 + SEARCH_TIMEOUT = 30 # 搜索超时时间（秒） + + # 处理配置 + PROCESS_INTERVAL = 300 # 处理间隔（秒） + BATCH_SIZE = 5 # 批量处理数量 + + # 日志配置 + LOG_DIR = os.path.join(os.path.dirname(__file__), 'logs') \ No newline at end of file diff --git a/data/param_auto.db b/data/param_auto.db new file mode 100644 index 0000000..aa4c50f Binary files /dev/null and b/data/param_auto.db differ diff --git a/logs/app.log b/logs/app.log new file mode 100644 index 0000000..5304c3d --- /dev/null +++ b/logs/app.log @@ -0,0 +1,34 @@ +2026-07-12 01:06:36,858 - param_auto_manager - INFO - 初始化数据库... +2026-07-12 01:06:36,859 - apscheduler.scheduler - INFO - Adding job tentatively -- it will be properly scheduled when the scheduler starts +2026-07-12 01:06:36,859 - scheduler - INFO - 添加定时任务: auto_process, 间隔: 300秒 +2026-07-12 01:06:36,860 - apscheduler.scheduler - INFO - Added job "auto_process_task" to job store "default" +2026-07-12 01:06:36,860 - apscheduler.scheduler - INFO - Scheduler started +2026-07-12 01:06:36,860 - scheduler - INFO - 定时任务调度器已启动 +2026-07-12 01:06:36,861 - param_auto_manager - INFO - 定时任务调度器已启动 +2026-07-12 01:06:36,861 - param_auto_manager - INFO - 启动服务，端口: 16043 + * Serving Flask app 'app' + * Debug mode: on +2026-07-12 01:06:36,878 - werkzeug - INFO - WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead. + * Running on all addresses (0.0.0.0) + * Running on http://127.0.0.1:16043 + * Running on http://192.168.0.101:16043 +2026-07-12 01:06:36,878 - werkzeug - INFO - Press CTRL+C to quit +2026-07-12 01:06:36,881 - werkzeug - INFO - * Restarting with stat +2026-07-12 01:06:37,478 - param_auto_manager - INFO - 初始化数据库... +2026-07-12 01:06:37,479 - apscheduler.scheduler - INFO - Adding job tentatively -- it will be properly scheduled when the scheduler starts +2026-07-12 01:06:37,479 - scheduler - INFO - 添加定时任务: auto_process, 间隔: 300秒 +2026-07-12 01:06:37,480 - apscheduler.scheduler - INFO - Added job "auto_process_task" to job store "default" +2026-07-12 01:06:37,480 - apscheduler.scheduler - INFO - Scheduler started +2026-07-12 01:06:37,481 - scheduler - INFO - 定时任务调度器已启动 +2026-07-12 01:06:37,481 - param_auto_manager - INFO - 定时任务调度器已启动 +2026-07-12 01:06:37,481 - param_auto_manager - INFO - 启动服务，端口: 16043 +2026-07-12 01:06:37,493 - werkzeug - WARNING - * Debugger is active! +2026-07-12 01:06:37,657 - werkzeug - INFO - * Debugger PIN: 854-773-964 +2026-07-12 01:06:40,247 - werkzeug - INFO - 127.0.0.1 - - [12/Jul/2026 01:06:40] "GET / HTTP/1.1" 200 - +2026-07-12 01:06:46,294 - werkzeug - INFO - 127.0.0.1 - - [12/Jul/2026 01:06:46] "GET /api/system/stats HTTP/1.1" 200 - +2026-07-12 01:06:46,302 - werkzeug - INFO - 127.0.0.1 - - [12/Jul/2026 01:06:46] "GET /api/products/pending HTTP/1.1" 200 - +2026-07-12 01:06:46,321 - werkzeug - INFO - 127.0.0.1 - - [12/Jul/2026 01:06:46] "GET /api/articles HTTP/1.1" 200 - +2026-07-12 01:06:56,298 - werkzeug - INFO - 127.0.0.1 - - [12/Jul/2026 01:06:56] "POST /api/articles HTTP/1.1" 200 - +2026-07-12 01:06:56,319 - werkzeug - INFO - 127.0.0.1 - - [12/Jul/2026 01:06:56] "POST /api/products/pending HTTP/1.1" 200 - +2026-07-12 01:06:59,746 - werkzeug - INFO - 127.0.0.1 - - [12/Jul/2026 01:06:59] "GET /api/articles/search?q=GPT-4o HTTP/1.1" 200 - +2026-07-12 01:06:59,757 - werkzeug - INFO - 127.0.0.1 - - [12/Jul/2026 01:06:59] "GET /api/products/pending HTTP/1.1" 200 - diff --git a/models/__pycache__/database.cpython-310.pyc b/models/__pycache__/database.cpython-310.pyc new file mode 100644 index 0000000..766fd05 Binary files /dev/null and b/models/__pycache__/database.cpython-310.pyc differ diff --git a/models/database.py b/models/database.py new file mode 100644 index 0000000..dc99791 --- /dev/null +++ b/models/database.py @@ -0,0 +1,313 @@ +""" +数据库模型和操作 +""" +import sqlite3 +import json +from datetime import datetime +from contextlib import contextmanager +from config import Config + +class Database: + def __init__(self, db_path=None): + self.db_path = db_path or Config.DATABASE + self.init_db() + + @contextmanager + def get_connection(self): + """获取数据库连接""" + conn = sqlite3.connect(self.db_path) + conn.row_factory = sqlite3.Row + try: + yield conn + finally: + conn.close() + + def init_db(self): + """初始化数据库""" + with self.get_connection() as conn: + cursor = conn.cursor() + + # 内容库表 + cursor.execute(''' + CREATE TABLE IF NOT EXISTS articles ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + product_names TEXT NOT NULL, + category TEXT, + keywords TEXT, + summary TEXT, + content TEXT, + source TEXT, + url TEXT, + fetch_date DATETIME DEFAULT CURRENT_TIMESTAMP, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME DEFAULT CURRENT_TIMESTAMP + ) + ''') + + # 待处理产品列表 + cursor.execute(''' + CREATE TABLE IF NOT EXISTS pending_products ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + product_name TEXT NOT NULL UNIQUE, + category TEXT, + subcategory TEXT, + priority INTEGER DEFAULT 0, + source TEXT, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME DEFAULT CURRENT_TIMESTAMP + ) + ''') + + # 处理中产品列表 + cursor.execute(''' + CREATE TABLE IF NOT EXISTS processing_products ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + product_name TEXT NOT NULL UNIQUE, + category TEXT, + subcategory TEXT, + status TEXT DEFAULT 'processing', + started_at DATETIME DEFAULT CURRENT_TIMESTAMP, + error_message TEXT + ) + ''') + + # 处理历史记录 + cursor.execute(''' + CREATE TABLE IF NOT EXISTS process_history ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + product_name TEXT NOT NULL, + category TEXT, + subcategory TEXT, + status TEXT, + review_id TEXT, + submitted_at DATETIME DEFAULT CURRENT_TIMESTAMP, + details TEXT + ) + ''') + + # 任务配置表 + cursor.execute(''' + CREATE TABLE IF NOT EXISTS task_configs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL UNIQUE, + config TEXT, + enabled INTEGER DEFAULT 1, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME DEFAULT CURRENT_TIMESTAMP + ) + ''') + + # 系统配置表 + cursor.execute(''' + CREATE TABLE IF NOT EXISTS system_config ( + key TEXT PRIMARY KEY, + value TEXT, + updated_at DATETIME DEFAULT CURRENT_TIMESTAMP + ) + ''') + + conn.commit() + +# ========== 内容库操作 ========== + def add_article(self, product_names, category, keywords, summary, content, source, url=None): + """添加文章到内容库""" + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute(''' + INSERT INTO articles (product_names, category, keywords, summary, content, source, url) + VALUES (?, ?, ?, ?, ?, ?, ?) + ''', (json.dumps(product_names, ensure_ascii=False), category, + json.dumps(keywords, ensure_ascii=False), summary, content, source, url)) + conn.commit() + return cursor.lastrowid + + def search_articles(self, keyword, category=None): + """搜索文章""" + with self.get_connection() as conn: + cursor = conn.cursor() + if category: + cursor.execute(''' + SELECT * FROM articles + WHERE (product_names LIKE ? OR keywords LIKE ? OR summary LIKE ? OR content LIKE ?) + AND category = ? + ORDER BY fetch_date DESC + ''', (f'%{keyword}%', f'%{keyword}%', f'%{keyword}%', f'%{keyword}%', category)) + else: + cursor.execute(''' + SELECT * FROM articles + WHERE product_names LIKE ? OR keywords LIKE ? OR summary LIKE ? OR content LIKE ? + ORDER BY fetch_date DESC + ''', (f'%{keyword}%', f'%{keyword}%', f'%{keyword}%', f'%{keyword}%')) + return [dict(row) for row in cursor.fetchall()] + + def get_article_by_id(self, article_id): + """获取文章详情""" + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute('SELECT * FROM articles WHERE id = ?', (article_id,)) + row = cursor.fetchone() + return dict(row) if row else None + + def get_all_articles(self, limit=100, offset=0): + """获取所有文章""" + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute('SELECT * FROM articles ORDER BY fetch_date DESC LIMIT ? OFFSET ?', (limit, offset)) + return [dict(row) for row in cursor.fetchall()] + + def delete_article(self, article_id): + """删除文章""" + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute('DELETE FROM articles WHERE id = ?', (article_id,)) + conn.commit() + return cursor.rowcount > 0 + +# ========== 待处理产品操作 ========== + def add_pending_product(self, product_name, category=None, subcategory=None, priority=0, source='manual'): + """添加待处理产品""" + with self.get_connection() as conn: + cursor = conn.cursor() + try: + cursor.execute(''' + INSERT INTO pending_products (product_name, category, subcategory, priority, source) + VALUES (?, ?, ?, ?, ?) + ''', (product_name, category, subcategory, priority, source)) + conn.commit() + return cursor.lastrowid + except sqlite3.IntegrityError: + # 产品已存在，更新优先级 + cursor.execute(''' + UPDATE pending_products + SET priority = MAX(priority, ?), updated_at = CURRENT_TIMESTAMP + WHERE product_name = ? + ''', (priority, product_name)) + conn.commit() + return None + + def get_pending_products(self, limit=10, order_by='priority'): + """获取待处理产品列表""" + with self.get_connection() as conn: + cursor = conn.cursor() + if order_by == 'priority': + cursor.execute('SELECT * FROM pending_products ORDER BY priority DESC, created_at ASC LIMIT ?', (limit,)) + else: + cursor.execute('SELECT * FROM pending_products ORDER BY created_at ASC LIMIT ?', (limit,)) + return [dict(row) for row in cursor.fetchall()] + + def get_pending_count(self): + """获取待处理产品数量""" + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute('SELECT COUNT(*) FROM pending_products') + return cursor.fetchone()[0] + + def remove_pending_product(self, product_name): + """从待处理列表移除产品""" + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute('DELETE FROM pending_products WHERE product_name = ?', (product_name,)) + conn.commit() + return cursor.rowcount > 0 + +# ========== 处理中产品操作 ========== + def start_processing(self, product_name, category, subcategory): + """开始处理产品""" + with self.get_connection() as conn: + cursor = conn.cursor() + try: + cursor.execute(''' + INSERT INTO processing_products (product_name, category, subcategory, status) + VALUES (?, ?, ?, 'processing') + ''', (product_name, category, subcategory)) + conn.commit() + return cursor.lastrowid + except sqlite3.IntegrityError: + return None + + def finish_processing(self, product_name, status='completed', error_message=None): + """完成处理""" + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute('DELETE FROM processing_products WHERE product_name = ?', (product_name,)) + conn.commit() + return cursor.rowcount > 0 + + def get_processing_products(self): + """获取处理中的产品""" + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute('SELECT * FROM processing_products') + return [dict(row) for row in cursor.fetchall()] + +# ========== 处理历史操作 ========== + def add_process_history(self, product_name, category, subcategory, status, review_id=None, details=None): + """添加处理历史""" + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute(''' + INSERT INTO process_history (product_name, category, subcategory, status, review_id, details) + VALUES (?, ?, ?, ?, ?, ?) + ''', (product_name, category, subcategory, status, review_id, + json.dumps(details, ensure_ascii=False) if details else None)) + conn.commit() + return cursor.lastrowid + + def get_process_history(self, limit=100): + """获取处理历史""" + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute('SELECT * FROM process_history ORDER BY submitted_at DESC LIMIT ?', (limit,)) + return [dict(row) for row in cursor.fetchall()] + + def get_history_by_product(self, product_name): + """获取指定产品的处理历史""" + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute('SELECT * FROM process_history WHERE product_name = ? ORDER BY submitted_at DESC', (product_name,)) + return [dict(row) for row in cursor.fetchall()] + +# ========== 任务配置操作 ========== + def save_task_config(self, name, config): + """保存任务配置""" + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute(''' + INSERT OR REPLACE INTO task_configs (name, config, updated_at) + VALUES (?, ?, CURRENT_TIMESTAMP) + ''', (name, json.dumps(config, ensure_ascii=False))) + conn.commit() + + def get_task_config(self, name): + """获取任务配置""" + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute('SELECT * FROM task_configs WHERE name = ?', (name,)) + row = cursor.fetchone() + if row: + result = dict(row) + result['config'] = json.loads(result['config']) + return result + return None + +# ========== 系统配置操作 ========== + def get_system_config(self, key, default=None): + """获取系统配置""" + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute('SELECT value FROM system_config WHERE key = ?', (key,)) + row = cursor.fetchone() + return row['value'] if row else default + + def set_system_config(self, key, value): + """设置系统配置""" + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute(''' + INSERT OR REPLACE INTO system_config (key, value, updated_at) + VALUES (?, ?, CURRENT_TIMESTAMP) + ''', (key, value)) + conn.commit() + +# 全局数据库实例 +db = Database() \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..7805951 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,8 @@ +flask==3.0.0 +flask-cors==4.0.0 +sqlite3 +requests==2.31.0 +beautifulsoup4==4.12.2 +lxml==4.9.3 +apscheduler==3.10.4 +python-dotenv==1.0.0 \ No newline at end of file diff --git a/routes/__pycache__/articles.cpython-310.pyc b/routes/__pycache__/articles.cpython-310.pyc new file mode 100644 index 0000000..f3e27c8 Binary files /dev/null and b/routes/__pycache__/articles.cpython-310.pyc differ diff --git a/routes/__pycache__/products.cpython-310.pyc b/routes/__pycache__/products.cpython-310.pyc new file mode 100644 index 0000000..6cbd34d Binary files /dev/null and b/routes/__pycache__/products.cpython-310.pyc differ diff --git a/routes/__pycache__/system.cpython-310.pyc b/routes/__pycache__/system.cpython-310.pyc new file mode 100644 index 0000000..3ee58e7 Binary files /dev/null and b/routes/__pycache__/system.cpython-310.pyc differ diff --git a/routes/articles.py b/routes/articles.py new file mode 100644 index 0000000..8244e31 --- /dev/null +++ b/routes/articles.py @@ -0,0 +1,135 @@ +""" +文章内容库管理 API +""" +from flask import Blueprint, request, jsonify +from models.database import db +from services.search_service import search_service + +bp = Blueprint('articles', __name__, url_prefix='/api/articles') + +@bp.route('', methods=['GET']) +def list_articles(): + """获取文章列表""" + limit = request.args.get('limit', 100, type=int) + offset = request.args.get('offset', 0, type=int) + + articles = db.get_all_articles(limit=limit, offset=offset) + + # 解析JSON字段 + for article in articles: + article['product_names'] = __import__('json').loads(article.get('product_names', '[]')) + article['keywords'] = __import__('json').loads(article.get('keywords', '[]')) + + return jsonify({ + 'success': True, + 'articles': articles, + 'count': len(articles) + }) + +@bp.route('/search', methods=['GET']) +def search_articles(): + """搜索文章""" + keyword = request.args.get('q', '') + category = request.args.get('category') + + if not keyword: + return jsonify({'error': '请提供搜索关键词'}), 400 + + articles = db.search_articles(keyword, category) + + # 解析JSON字段 + for article in articles: + article['product_names'] = __import__('json').loads(article.get('product_names', '[]')) + article['keywords'] = __import__('json').loads(article.get('keywords', '[]')) + + return jsonify({ + 'success': True, + 'articles': articles, + 'count': len(articles) + }) + +@bp.route('/', methods=['GET']) +def get_article(article_id): + """获取文章详情""" + article = db.get_article_by_id(article_id) + + if not article: + return jsonify({'error': '文章不存在'}), 404 + + article['product_names'] = __import__('json').loads(article.get('product_names', '[]')) + article['keywords'] = __import__('json').loads(article.get('keywords', '[]')) + + return jsonify({ + 'success': True, + 'article': article + }) + +@bp.route('', methods=['POST']) +def create_article(): + """创建文章（手动添加）""" + data = request.get_json() + + required_fields = ['product_names', 'summary', 'content', 'source'] + for field in required_fields: + if field not in data: + return jsonify({'error': f'缺少必填字段: {field}'}), 400 + + article_id = search_service.save_to_articles( + product_names=data['product_names'], + category=data.get('category'), + keywords=data.get('keywords', []), + summary=data['summary'], + content=data['content'], + source=data['source'], + url=data.get('url') + ) + + return jsonify({ + 'success': True, + 'article_id': article_id, + 'message': '文章创建成功' + }) + +@bp.route('/', methods=['DELETE']) +def delete_article(article_id): + """删除文章""" + success = db.delete_article(article_id) + + if success: + return jsonify({ + 'success': True, + 'message': '文章已删除' + }) + else: + return jsonify({'error': '文章不存在或删除失败'}), 404 + +@bp.route('/fetch', methods=['POST']) +def fetch_article(): + """从URL抓取文章""" + data = request.get_json() + url = data.get('url') + + if not url: + return jsonify({'error': '请提供URL'}), 400 + + result = search_service.fetch_url_content(url) + + if result: + # 自动保存到内容库 + article_id = search_service.save_to_articles( + product_names=data.get('product_names', [result['title']]), + category=data.get('category'), + keywords=data.get('keywords', []), + summary=result.get('description', ''), + content=result['content'], + source=url, + url=url + ) + + return jsonify({ + 'success': True, + 'article_id': article_id, + 'data': result + }) + else: + return jsonify({'error': '抓取失败'}), 500 \ No newline at end of file diff --git a/routes/products.py b/routes/products.py new file mode 100644 index 0000000..293b133 --- /dev/null +++ b/routes/products.py @@ -0,0 +1,216 @@ +""" +产品处理 API +""" +from flask import Blueprint, request, jsonify +from models.database import db +from services.process_service import process_service + +bp = Blueprint('products', __name__, url_prefix='/api/products') + +@bp.route('/pending', methods=['GET']) +def list_pending(): + """获取待处理产品列表""" + limit = request.args.get('limit', 20, type=int) + order_by = request.args.get('order_by', 'priority') + + products = db.get_pending_products(limit=limit, order_by=order_by) + count = db.get_pending_count() + + return jsonify({ + 'success': True, + 'products': products, + 'count': count + }) + +@bp.route('/pending', methods=['POST']) +def add_pending(): + """添加待处理产品""" + data = request.get_json() + + if isinstance(data, dict): + products = [data] + elif isinstance(data, list): + products = data + else: + return jsonify({'error': '无效的数据格式'}), 400 + + added_count = 0 + for item in products: + if 'product_name' not in item: + continue + + result = db.add_pending_product( + product_name=item['product_name'], + category=item.get('category'), + subcategory=item.get('subcategory'), + priority=item.get('priority', 0), + source=item.get('source', 'manual') + ) + if result: + added_count += 1 + + return jsonify({ + 'success': True, + 'added_count': added_count, + 'message': f'成功添加 {added_count} 个产品到待处理列表' + }) + +@bp.route('/pending/', methods=['DELETE']) +def remove_pending(product_name): + """从待处理列表移除产品""" + success = db.remove_pending_product(product_name) + + if success: + return jsonify({ + 'success': True, + 'message': '产品已从待处理列表移除' + }) + else: + return jsonify({'error': '产品不存在'}), 404 + +@bp.route('/processing', methods=['GET']) +def list_processing(): + """获取正在处理的产品列表""" + products = db.get_processing_products() + + return jsonify({ + 'success': True, + 'products': products, + 'count': len(products) + }) + +@bp.route('/history', methods=['GET']) +def list_history(): + """获取处理历史""" + limit = request.args.get('limit', 100, type=int) + history = db.get_process_history(limit=limit) + + # 解析JSON字段 + for item in history: + if item.get('details'): + item['details'] = __import__('json').loads(item['details']) + + return jsonify({ + 'success': True, + 'history': history, + 'count': len(history) + }) + +@bp.route('/history/', methods=['GET']) +def get_product_history(product_name): + """获取指定产品的处理历史""" + history = db.get_history_by_product(product_name) + + for item in history: + if item.get('details'): + item['details'] = __import__('json').loads(item['details']) + + return jsonify({ + 'success': True, + 'product_name': product_name, + 'history': history, + 'count': len(history) + }) + +@bp.route('/process', methods=['POST']) +def process_single(): + """处理单个产品""" + data = request.get_json() + + if 'product_name' not in data: + return jsonify({'error': '请提供产品名称'}), 400 + + product_info = { + 'product_name': data['product_name'], + 'category': data.get('category'), + 'subcategory': data.get('subcategory') + } + + # 检查是否正在处理 + processing = db.get_processing_products() + if any(p['product_name'] == product_info['product_name'] for p in processing): + return jsonify({'error': '该产品正在处理中'}), 400 + + # 添加到处理中列表 + db.start_processing( + product_name=product_info['product_name'], + category=product_info['category'], + subcategory=product_info['subcategory'] + ) + + try: + # 执行处理 + result = process_service.process_product(product_info) + + # 从待处理列表移除 + db.remove_pending_product(product_info['product_name']) + + # 如果发现新产品，已在process_service中添加到待处理列表 + + return jsonify({ + 'success': result['success'], + 'message': result['message'], + 'review_id': result.get('review_id'), + 'new_products': result.get('new_products', []) + }) + finally: + # 完成处理，从处理中列表移除 + db.finish_processing(product_info['product_name']) + +@bp.route('/process/batch', methods=['POST']) +def process_batch(): + """批量处理产品""" + data = request.get_json() + limit = data.get('limit', 5) + + # 获取待处理产品 + products = db.get_pending_products(limit=limit) + + if not products: + return jsonify({ + 'success': True, + 'message': '没有待处理的产品', + 'processed': 0 + }) + + results = [] + for product in products: + # 检查是否正在处理 + processing = db.get_processing_products() + if any(p['product_name'] == product['product_name'] for p in processing): + results.append({ + 'product_name': product['product_name'], + 'success': False, + 'message': '正在处理中' + }) + continue + + # 添加到处理中列表 + db.start_processing( + product_name=product['product_name'], + category=product.get('category'), + subcategory=product.get('subcategory') + ) + + try: + # 执行处理 + result = process_service.process_product(product) + + # 从待处理列表移除 + db.remove_pending_product(product['product_name']) + + results.append({ + 'product_name': product['product_name'], + 'success': result['success'], + 'message': result['message'], + 'review_id': result.get('review_id') + }) + finally: + # 完成处理 + db.finish_processing(product['product_name']) + + return jsonify({ + 'success': True, + 'processed': len(results), + 'results': results + }) \ No newline at end of file diff --git a/routes/system.py b/routes/system.py new file mode 100644 index 0000000..8c08fa2 --- /dev/null +++ b/routes/system.py @@ -0,0 +1,68 @@ +""" +系统管理 API +""" +from flask import Blueprint, request, jsonify +from models.database import db + +bp = Blueprint('system', __name__, url_prefix='/api/system') + +@bp.route('/config', methods=['GET']) +def get_config(): + """获取系统配置""" + configs = { + 'auto_process_enabled': db.get_system_config('auto_process_enabled', 'true'), + 'process_interval': db.get_system_config('process_interval', '300'), + 'batch_size': db.get_system_config('batch_size', '5') + } + + return jsonify({ + 'success': True, + 'config': configs + }) + +@bp.route('/config', methods=['PUT']) +def update_config(): + """更新系统配置""" + data = request.get_json() + + for key, value in data.items(): + db.set_system_config(key, str(value)) + + return jsonify({ + 'success': True, + 'message': '配置已更新' + }) + +@bp.route('/stats', methods=['GET']) +def get_stats(): + """获取系统统计信息""" + pending_count = db.get_pending_count() + processing_count = len(db.get_processing_products()) + + # 获取最近处理历史 + recent_history = db.get_process_history(limit=10) + success_count = len([h for h in recent_history if h['status'] == 'submitted']) + failed_count = len([h for h in recent_history if h['status'] in ['failed', 'error']]) + + # 获取内容库统计 + articles = db.get_all_articles(limit=1000) + + return jsonify({ + 'success': True, + 'stats': { + 'pending_products': pending_count, + 'processing_products': processing_count, + 'recent_success': success_count, + 'recent_failed': failed_count, + 'total_articles': len(articles) + } + }) + +@bp.route('/health', methods=['GET']) +def health_check(): + """健康检查""" + return jsonify({ + 'success': True, + 'status': 'healthy', + 'message': '系统运行正常' + }) \ No newline at end of file diff --git a/services/__pycache__/paramhub_client.cpython-310.pyc b/services/__pycache__/paramhub_client.cpython-310.pyc new file mode 100644 index 0000000..b40381d Binary files /dev/null and b/services/__pycache__/paramhub_client.cpython-310.pyc differ diff --git a/services/__pycache__/process_service.cpython-310.pyc b/services/__pycache__/process_service.cpython-310.pyc new file mode 100644 index 0000000..b3bdc49 Binary files /dev/null and b/services/__pycache__/process_service.cpython-310.pyc differ diff --git a/services/__pycache__/search_service.cpython-310.pyc b/services/__pycache__/search_service.cpython-310.pyc new file mode 100644 index 0000000..dd90e38 Binary files /dev/null and b/services/__pycache__/search_service.cpython-310.pyc differ diff --git a/services/paramhub_client.py b/services/paramhub_client.py new file mode 100644 index 0000000..7541219 --- /dev/null +++ b/services/paramhub_client.py @@ -0,0 +1,132 @@ +""" +ParamHub API 客户端服务 +""" +import requests +from config import Config + +class ParamHubClient: + def __init__(self): + self.base_url = Config.PARAMHUB_BASE_URL + self.password = Config.PARAMHUB_PASSWORD + self.session = None + + def login(self): + """登录获取session""" + try: + self.session = requests.Session() + response = self.session.post( + f'{self.base_url}/login', + json={'password': self.password} + ) + return response.json().get('success', False) + except Exception as e: + print(f"登录失败: {str(e)}") + return False + + def get_categories(self): + """获取所有分类""" + try: + if not self.session: + self.login() + + response = self.session.get(f'{self.base_url}/api/categories?all=1') + return response.json() + except Exception as e: + print(f"获取分类失败: {str(e)}") + return [] + + def get_category_fields(self, category_id): + """获取分类的字段配置""" + try: + if not self.session: + self.login() + + response = self.session.get(f'{self.base_url}/api/categories/{category_id}') + return response.json() + except Exception as e: + print(f"获取分类字段失败: {str(e)}") + return None + + def submit_for_review(self, category_type, data, category_id=None): + """ + 提交数据到待审核区 + + Args: + category_type: 分类类型 (model/gpu/cpu/dynamic) + data: 产品数据 + category_id: 分类ID（用于动态分类） + + Returns: + (success, review_id or error_message) + """ + try: + if not self.session: + self.login() + + # 根据分类类型选择API端点 + if category_type == 'model': + endpoint = f'{self.base_url}/api/models' + elif category_type == 'gpu': + endpoint = f'{self.base_url}/api/gpus' + elif category_type == 'cpu': + endpoint = f'{self.base_url}/api/cpus' + elif category_type == 'dynamic' and category_id: + endpoint = f'{self.base_url}/api/items/{category_id}' + else: + return False, "无效的分类类型或缺少分类ID" + + # 添加审核模式需要的字段 + data['status'] = 'pending' + + response = self.session.post(endpoint, json=data) + result = response.json() + + if response.status_code == 200 or response.status_code == 201: + return True, result.get('review_id', 'submitted') + else: + return False, result.get('error', '提交失败') + except Exception as e: + return False, str(e) + + def get_reviews(self, status='pending'): + """获取待审核列表""" + try: + if not self.session: + self.login() + + response = self.session.get(f'{self.base_url}/api/reviews?status={status}') + return response.json() + except Exception as e: + print(f"获取审核列表失败: {str(e)}") + return [] + + def get_review_count(self): + """获取待审核数量""" + try: + if not self.session: + self.login() + + response = self.session.get(f'{self.base_url}/api/reviews/count') + return response.json().get('count', 0) + except Exception as e: + print(f"获取审核数量失败: {str(e)}") + return 0 + + def send_notification(self, message): + """发送通知到后台管理""" + try: + if not self.session: + self.login() + + # 使用通知API发送通知 + response = self.session.post( + f'{self.base_url}/api/notifications', + json={'message': message} + ) + return response.status_code == 200 or response.status_code == 201 + except Exception as e: + print(f"发送通知失败: {str(e)}") + return False + +# 全局客户端实例 +paramhub_client = ParamHubClient() \ No newline at end of file diff --git a/services/process_service.py b/services/process_service.py new file mode 100644 index 0000000..c8e1a8a --- /dev/null +++ b/services/process_service.py @@ -0,0 +1,412 @@ +""" +数据处理服务 - 核心处理逻辑 +""" +import json +import re +from datetime import datetime +from config import Config +from models.database import db +from services.search_service import search_service +from services.paramhub_client import paramhub_client + +class DataProcessService: + def __init__(self): + self.config = Config + + def process_product(self, product_info): + """ + 处理单个产品的完整流程 + + Args: + product_info: { + 'product_name': str, + 'category': str (可选), + 'subcategory': str (可选) + } + + Returns: + { + 'success': bool, + 'message': str, + 'review_id': str (如果成功提交), + 'new_products': list (发现的新产品) + } + """ + product_name = product_info['product_name'] + category = product_info.get('category') + subcategory = product_info.get('subcategory') + + result = { + 'success': False, + 'message': '', + 'review_id': None, + 'new_products': [] + } + + try: + # 1. 从内容库和互联网搜索原始数据 + print(f"[处理] 开始处理产品: {product_name}") + search_results = search_service.search_all( + keyword=product_name, + category=category, + include_internet=True + ) + + if search_results['total'] == 0: + # 没有找到数据，发送通知 + paramhub_client.send_notification(f"未找到产品 '{product_name}' 的相关数据") + result['message'] = '未找到相关数据' + return result + + # 2. 提取对应产品的具体内容（排除无关产品） + extracted_data = self.extract_product_data( + product_name, + search_results, + category, + subcategory + ) + + if not extracted_data: + result['message'] = '无法提取有效数据' + return result + + # 3. 按照类别和子类别字段填充内容 + filled_data = self.fill_product_fields( + extracted_data, + category, + subcategory + ) + + if not filled_data: + result['message'] = '填充数据失败' + return result + + # 4. 提交到待审核区 + category_type = self.get_category_type(category) + success, review_id_or_error = paramhub_client.submit_for_review( + category_type, + filled_data, + subcategory + ) + + if success: + # 记录处理历史 + db.add_process_history( + product_name=product_name, + category=category, + subcategory=subcategory, + status='submitted', + review_id=review_id_or_error, + details={'data': filled_data} + ) + + result['success'] = True + result['message'] = f'已提交审核，review_id: {review_id_or_error}' + result['review_id'] = review_id_or_error + + print(f"[处理] 产品 {product_name} 提交成功") + else: + result['message'] = f'提交失败: {review_id_or_error}' + db.add_process_history( + product_name=product_name, + category=category, + subcategory=subcategory, + status='failed', + details={'error': review_id_or_error} + ) + + # 5. 检查是否发现新的未处理产品 + new_products = self.discover_new_products( + product_name, + search_results, + category + ) + result['new_products'] = new_products + + return result + + except Exception as e: + error_msg = f"处理失败: {str(e)}" + print(f"[错误] {error_msg}") + result['message'] = error_msg + db.add_process_history( + product_name=product_name, + category=category, + subcategory=subcategory, + status='error', + details={'error': str(e)} + ) + return result + + def extract_product_data(self, target_product, search_results, category, subcategory): + """ + 从搜索结果中提取目标产品的具体内容 + 排除无关产品 + """ + extracted = { + 'name': target_product, + 'category': category, + 'subcategory': subcategory, + 'raw_data': [] + } + + # 收集所有相关内容 + all_content = [] + + # 从内容库结果中提取 + for article in search_results.get('articles', []): + product_names = json.loads(article.get('product_names', '[]')) + + # 检查是否包含目标产品 + if self.is_product_match(target_product, product_names): + all_content.append({ + 'source': article.get('source'), + 'url': article.get('url'), + 'summary': article.get('summary'), + 'content': article.get('content'), + 'keywords': json.loads(article.get('keywords', '[]')) + }) + + # 从互联网搜索结果中提取 + for item in search_results.get('internet', []): + if self.is_product_match(target_product, [item.get('title', '')]): + all_content.append({ + 'source': 'internet', + 'url': item.get('url'), + 'content': item.get('content') + }) + + if not all_content: + return None + + extracted['raw_data'] = all_content + return extracted + + def fill_product_fields(self, extracted_data, category, subcategory): + """ + 根据分类字段配置，填充产品数据 + 严格按照来源数据，不创造内容 + """ + if not extracted_data: + return None + + # 获取分类字段配置 + category_info = paramhub_client.get_category_fields(subcategory) if subcategory else None + + # 基础字段 + filled_data = { + 'name': extracted_data['name'], + 'visible': True, + 'is_pinned': False + } + + # 根据分类类型填充字段 + category_type = self.get_category_type(category) + + if category_type == 'model': + filled_data.update(self.extract_model_fields(extracted_data['raw_data'])) + elif category_type == 'gpu': + filled_data.update(self.extract_gpu_fields(extracted_data['raw_data'])) + elif category_type == 'cpu': + filled_data.update(self.extract_cpu_fields(extracted_data['raw_data'])) + else: + # 动态分类，使用分类字段配置 + if category_info and 'fields' in category_info: + for field in category_info['fields']: + value = self.extract_field_from_data(field, extracted_data['raw_data']) + if value: + filled_data[field] = value + + # 添加数据来源 + filled_data['_source'] = 'auto_manager' + filled_data['_extracted_at'] = datetime.now().isoformat() + + return filled_data + + def extract_model_fields(self, raw_data): + """提取AI模型字段""" + fields = {} + + for data in raw_data: + content = data.get('content', '') or '' + summary = data.get('summary', '') or '' + text = f"{summary}\n{content}" + + # 提取参数量 + if 'parameters' not in fields: + params_match = re.search(r'(\d+(?:\.\d+)?)\s*[Bb]', text) + if params_match: + fields['parameters'] = f"{params_match.group(1)}B" + + # 提取上下文长度 + if 'context_length' not in fields: + ctx_match = re.search(r'context[:\s]+(\d+)', text, re.I) + if ctx_match: + fields['context_length'] = int(ctx_match.group(1)) + + # 提取组织 + if 'organization' not in fields: + org_match = re.search(r'(?:by\s+|from\s+|developed\s+by\s+)([\w\s]+?)(?:\s|,|\.|$)', text, re.I) + if org_match: + fields['organization'] = org_match.group(1).strip() + + # 提取发布日期 + if 'publish_date' not in fields: + date_match = re.search(r'(\d{4}[-/]\d{1,2}[-/]\d{1,2})', text) + if date_match: + fields['publish_date'] = date_match.group(1).replace('/', '-') + + return fields + + def extract_gpu_fields(self, raw_data): + """提取GPU字段""" + fields = {} + + for data in raw_data: + content = data.get('content', '') or '' + summary = data.get('summary', '') or '' + text = f"{summary}\n{content}" + + # 提取显存 + if 'memory_gb' not in fields: + mem_match = re.search(r'(\d+)\s*GB', text) + if mem_match: + fields['memory_gb'] = int(mem_match.group(1)) + + # 提取CUDA核心 + if 'cuda_cores' not in fields: + cuda_match = re.search(r'(\d+)\s*(?:CUDA|cuda)\s*(?:cores|core)', text, re.I) + if cuda_match: + fields['cuda_cores'] = int(cuda_match.group(1)) + + # 提取价格 + if 'price_usd' not in fields: + price_match = re.search(r'\$(\d+(?:,\d+)*)', text) + if price_match: + fields['price_usd'] = int(price_match.group(1).replace(',', '')) + + return fields + + def extract_cpu_fields(self, raw_data): + """提取CPU字段""" + fields = {} + + for data in raw_data: + content = data.get('content', '') or '' + summary = data.get('summary', '') or '' + text = f"{summary}\n{content}" + + # 提取核心数 + if 'cores' not in fields: + cores_match = re.search(r'(\d+)\s*(?:cores?|Cores?)', text, re.I) + if cores_match: + fields['cores'] = int(cores_match.group(1)) + + # 提取线程数 + if 'threads' not in fields: + threads_match = re.search(r'(\d+)\s*(?:threads?|Threads?)', text, re.I) + if threads_match: + fields['threads'] = int(threads_match.group(1)) + + # 提取频率 + if 'base_clock' not in fields: + clock_match = re.search(r'(\d+(?:\.\d+)?)\s*GHz', text) + if clock_match: + fields['base_clock'] = float(clock_match.group(1)) + + return fields + + def extract_field_from_data(self, field_name, raw_data): + """从数据中提取指定字段""" + # 通用字段提取逻辑 + for data in raw_data: + content = data.get('content', '') or '' + summary = data.get('summary', '') or '' + text = f"{summary}\n{content}" + + # 尝试直接匹配字段名 + pattern = rf'{field_name}[:\s]+([^\n]+)' + match = re.search(pattern, text, re.I) + if match: + return match.group(1).strip() + + return None + + def get_category_type(self, category): + """获取分类类型""" + if not category: + return 'dynamic' + + category_lower = category.lower() + if 'model' in category_lower or 'ai' in category_lower or 'llm' in category_lower: + return 'model' + elif 'gpu' in category_lower: + return 'gpu' + elif 'cpu' in category_lower: + return 'cpu' + else: + return 'dynamic' + + def is_product_match(self, target_product, product_names): + """检查产品名称是否匹配""" + target = target_product.lower().strip() + + for name in product_names: + name_lower = name.lower().strip() + + # 完全匹配 + if target == name_lower: + return True + + # 包含匹配 + if target in name_lower or name_lower in target: + return True + + # 关键词匹配（去掉型号后缀） + target_base = re.sub(r'[-\d]+$', '', target) + name_base = re.sub(r'[-\d]+$', '', name_lower) + if target_base == name_base: + return True + + return False + + def discover_new_products(self, current_product, search_results, category): + """ + 从搜索结果中发现新的未处理产品 + """ + new_products = [] + + # 收集所有产品名称 + all_product_names = set() + + for article in search_results.get('articles', []): + product_names = json.loads(article.get('product_names', '[]')) + all_product_names.update(product_names) + + # 检查是否在待处理列表中 + for product_name in all_product_names: + # 排除当前产品 + if product_name.lower() == current_product.lower(): + continue + + # 检查是否已在待处理列表或已处理 + pending_count = db.get_pending_count() + pending_products = [p['product_name'] for p in db.get_pending_products(limit=1000)] + + if product_name not in pending_products: + # 检查是否已处理过 + history = db.get_history_by_product(product_name) + if not history: + # 添加到待处理列表 + db.add_pending_product( + product_name=product_name, + category=category, + source='discovered', + priority=1 # 发现的产品优先级较低 + ) + new_products.append(product_name) + + return new_products + +# 全局处理服务实例 +process_service = DataProcessService() \ No newline at end of file diff --git a/services/search_service.py b/services/search_service.py new file mode 100644 index 0000000..a310bd0 --- /dev/null +++ b/services/search_service.py @@ -0,0 +1,99 @@ +""" +搜索服务 - 从内容库和互联网搜索数据 +""" +import requests +from bs4 import BeautifulSoup +import json +from datetime import datetime +from config import Config +from models.database import db + +class SearchService: + def __init__(self): + self.timeout = Config.SEARCH_TIMEOUT + self.max_results = Config.SEARCH_MAX_RESULTS + + def search_internet(self, keyword, max_results=None): + """ + 从互联网搜索（使用搜索API或爬虫） + 这里暂时使用简单的搜索模拟 + """ + max_results = max_results or self.max_results + + # TODO: 接入真实的搜索API（如Google Custom Search、Bing等） + # 这里先返回空列表，等待后续接入真实API + results = [] + + return results + + def fetch_url_content(self, url): + """抓取网页内容""" + try: + headers = { + 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' + } + response = requests.get(url, headers=headers, timeout=self.timeout) + response.raise_for_status() + + soup = BeautifulSoup(response.text, 'lxml') + + # 提取标题 + title = soup.find('title') + title = title.text.strip() if title else '' + + # 提取正文（简单提取，可优化） + # 移除脚本和样式 + for script in soup(['script', 'style']): + script.decompose() + + # 提取文本 + text = soup.get_text(separator='\n', strip=True) + + # 提取元数据 + meta_desc = soup.find('meta', attrs={'name': 'description'}) + description = meta_desc['content'] if meta_desc else '' + + return { + 'title': title, + 'description': description, + 'content': text, + 'url': url, + 'fetch_date': datetime.now().isoformat() + } + except Exception as e: + print(f"抓取URL失败: {url}, 错误: {str(e)}") + return None + + def search_articles(self, keyword, category=None): + """从内容库搜索""" + return db.search_articles(keyword, category) + + def search_all(self, keyword, category=None, include_internet=True): + """ + 综合搜索：内容库 + 互联网 + """ + results = { + 'articles': [], + 'internet': [], + 'total': 0 + } + + # 1. 从内容库搜索 + articles = self.search_articles(keyword, category) + results['articles'] = articles + + # 2. 从互联网搜索（如果启用） + if include_internet: + internet_results = self.search_internet(keyword) + results['internet'] = internet_results + + results['total'] = len(articles) + len(results['internet']) + + return results + + def save_to_articles(self, product_names, category, keywords, summary, content, source, url=None): + """保存搜索结果到内容库""" + return db.add_article(product_names, category, keywords, summary, content, source, url) + +# 全局搜索服务实例 +search_service = SearchService() \ No newline at end of file diff --git a/start.sh b/start.sh new file mode 100755 index 0000000..9c911f4 --- /dev/null +++ b/start.sh @@ -0,0 +1,35 @@ +#!/bin/bash + +# 参数数据自动化管理系统启动脚本 + +APP_DIR="/home/openclaw/.openclaw/workspace-hz4th_coder/works/param-auto-manager" +LOG_DIR="$APP_DIR/logs" +PID_FILE="$APP_DIR/app.pid" + +cd $APP_DIR + +# 创建必要的目录 +mkdir -p $LOG_DIR +mkdir -p $APP_DIR/data + +# 检查是否已运行 +if [ -f "$PID_FILE" ]; then + PID=$(cat $PID_FILE) + if ps -p $PID > /dev/null 2>&1; then + echo "服务已在运行中 (PID: $PID)" + exit 1 + else + rm -f $PID_FILE + fi +fi + +# 设置Python路径 +export PYTHONPATH="$HOME/.local/lib/python3.12/site-packages:$PYTHONPATH" + +# 启动服务 +echo "启动参数数据自动化管理系统..." +nohup python3 app.py > $LOG_DIR/app.log 2>&1 & +echo $! > $PID_FILE + +echo "服务已启动 (PID: $(cat $PID_FILE))" +echo "日志文件: $LOG_DIR/app.log" \ No newline at end of file diff --git a/stop.sh b/stop.sh new file mode 100644 index 0000000..96ea199 --- /dev/null +++ b/stop.sh @@ -0,0 +1,38 @@ +#!/bin/bash + +# 停止脚本 + +APP_DIR="/home/openclaw/.openclaw/workspace-hz4th_coder/works/param-auto-manager" +PID_FILE="$APP_DIR/app.pid" + +if [ ! -f "$PID_FILE" ]; then + echo "PID文件不存在，服务可能未运行" + exit 1 +fi + +PID=$(cat $PID_FILE) + +if ps -p $PID > /dev/null 2>&1; then + echo "停止服务 (PID: $PID)..." + kill $PID + + # 等待进程结束 + for i in {1..10}; do + if ! ps -p $PID > /dev/null 2>&1; then + break + fi + sleep 1 + done + + # 如果进程还在运行，强制杀掉 + if ps -p $PID > /dev/null 2>&1; then + echo "强制停止服务..." + kill -9 $PID + fi + + echo "服务已停止" +else + echo "服务未运行" +fi + +rm -f $PID_FILE \ No newline at end of file diff --git a/utils/__pycache__/scheduler.cpython-310.pyc b/utils/__pycache__/scheduler.cpython-310.pyc new file mode 100644 index 0000000..900ac8d Binary files /dev/null and b/utils/__pycache__/scheduler.cpython-310.pyc differ diff --git a/utils/scheduler.py b/utils/scheduler.py new file mode 100644 index 0000000..2abf89a --- /dev/null +++ b/utils/scheduler.py @@ -0,0 +1,127 @@ +""" +定时任务调度器 +""" +from apscheduler.schedulers.background import BackgroundScheduler +from apscheduler.triggers.interval import IntervalTrigger +from datetime import datetime +from models.database import db +from services.process_service import process_service +from config import Config +import logging + +# 配置日志 +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger('scheduler') + +class TaskScheduler: + def __init__(self): + self.scheduler = BackgroundScheduler() + self.running = False + + def start(self): + """启动调度器""" + if not self.running: + self.scheduler.start() + self.running = True + logger.info("定时任务调度器已启动") + + def stop(self): + """停止调度器""" + if self.running: + self.scheduler.shutdown() + self.running = False + logger.info("定时任务调度器已停止") + + def add_interval_job(self, func, interval_seconds, job_id, **kwargs): + """添加定时任务""" + self.scheduler.add_job( + func, + trigger=IntervalTrigger(seconds=interval_seconds), + id=job_id, + replace_existing=True, + **kwargs + ) + logger.info(f"添加定时任务: {job_id}, 间隔: {interval_seconds}秒") + + def remove_job(self, job_id): + """移除定时任务""" + try: + self.scheduler.remove_job(job_id) + logger.info(f"移除定时任务: {job_id}") + except Exception as e: + logger.error(f"移除任务失败: {job_id}, 错误: {str(e)}") + + def get_jobs(self): + """获取所有任务""" + return self.scheduler.get_jobs() + +# 创建全局调度器实例 +task_scheduler = TaskScheduler() + +def auto_process_task(): + """ + 自动处理产品的定时任务 + """ + try: + # 检查是否启用自动处理 + enabled = db.get_system_config('auto_process_enabled', 'true') + if enabled.lower() != 'true': + logger.info("自动处理已禁用，跳过本次执行") + return + + # 获取批量处理数量 + batch_size = int(db.get_system_config('batch_size', '5')) + + # 获取待处理产品 + products = db.get_pending_products(limit=batch_size) + + if not products: + logger.info("没有待处理的产品") + return + + logger.info(f"开始处理 {len(products)} 个产品") + + for product in products: + try: + # 检查是否正在处理 + processing = db.get_processing_products() + if any(p['product_name'] == product['product_name'] for p in processing): + logger.warning(f"产品 {product['product_name']} 正在处理中，跳过") + continue + + # 添加到处理中列表 + db.start_processing( + product_name=product['product_name'], + category=product.get('category'), + subcategory=product.get('subcategory') + ) + + # 执行处理 + result = process_service.process_product(product) + + # 从待处理列表移除 + db.remove_pending_product(product['product_name']) + + logger.info(f"产品 {product['product_name']} 处理完成: {result['message']}") + + except Exception as e: + logger.error(f"处理产品 {product['product_name']} 时出错: {str(e)}") + db.finish_processing(product['product_name']) + finally: + # 确保从处理中列表移除 + db.finish_processing(product['product_name']) + + except Exception as e: + logger.error(f"自动处理任务执行失败: {str(e)}") + +def setup_auto_process_job(): + """设置自动处理定时任务""" + interval = int(db.get_system_config('process_interval', str(Config.PROCESS_INTERVAL))) + task_scheduler.add_interval_job( + auto_process_task, + interval_seconds=interval, + job_id='auto_process' + ) \ No newline at end of file