深圳设计网站排名缠绕机东莞网站建设技术支持

深圳设计网站排名,缠绕机东莞网站建设技术支持,nodejs同时做网站和后台管理,crm系统哪种品牌的好项目摘要名称#xff1a;多子项目协同中枢系统设计与实现项目背景与目标在当今复杂的软件开发环境中#xff0c;多项目协同、动态重构和灵活配置已成为核心需求。传统的单一架构系统难以满足快速变化的业务需求#xff0c;尤其是在需要频繁调整规则、生成新应用并管理其生命…项目摘要名称多子项目协同中枢系统设计与实现项目背景与目标在当今复杂的软件开发环境中多项目协同、动态重构和灵活配置已成为核心需求。传统的单一架构系统难以满足快速变化的业务需求尤其是在需要频繁调整规则、生成新应用并管理其生命周期的场景下。本项目旨在构建一个多子项目协同中枢实现信息系统架构语言文档抽象层与指令集架构程序文件逻辑层的双架构融合设计。系统通过物理层规约库文本将映射固化在本地数据中心形成一个完整的三层架构系统。核心设计理念1. 双架构融合设计信息系统架构语言文档抽象层负责规则定义、知识表示和语义理解指令集架构程序文件逻辑层负责具体实现、代码生成和执行控制物理层规约库文本作为连接两层架构的桥梁将抽象规则映射到具体实现2. 动态重构的备忘录仪表盘备忘录仪表盘是系统的核心组件具有以下特性协同看板记录执行历史、结果和当前进度位置协同指挥棒控制节拍和前进速度双重功能同时支持做笔记变量约束和记日志目标绑定动态适应随人类行为而动重定位、重配置、重命名等3. 三层架构体系规则定义层第一套定义生成应用的规则和标准应用生成层第二套基于规则自动生成可运行应用应用运行层第三套启动、停止和使用生成的应用系统架构与功能规则定义层核心功能规则注册与管理支持动态添加、更新和删除规则规则校验验证规则的合法性和有效性规则查询支持按分类、状态、优先级等条件筛选规则条件评估根据上下文动态评估规则条件技术实现基于Python的规则处理核心库RESTful API接口响应式Web界面JSON数据持久化应用生成层核心功能代码自动生成基于规则和模板生成可运行的应用代码应用部署自动创建应用目录结构和启动脚本应用管理支持应用的查询、删除和导出多模板支持提供不同类型的应用模板技术实现模板引擎和代码生成器动态端口分配应用生命周期管理可视化生成界面应用运行层核心功能应用启停管理应用的启动和停止状态监控实时监控应用的运行状态健康检查定期检查应用的健康状况日志管理收集和展示应用日志技术实现进程管理和监控健康检查机制实时日志收集交互式管理界面技术特色与创新点1. 纯原生技术栈后端纯原生Python无第三方依赖Python 3.6前端原生HTML/JS/CSS数据存储JSON文件通信HTTP服务2. 动态重构机制支持规则的动态更新和重构应用代码的自动生成和部署运行时的状态监控和调整3. 可视化管理界面响应式设计支持移动端访问直观的操作界面和实时反馈丰富的统计信息和监控数据4. 模块化设计清晰的层次结构和模块划分松耦合的组件设计易于扩展和维护应用场景1. 快速原型开发快速生成和部署原型应用支持多种应用类型便于演示和测试2. 多项目协同管理集中管理多个子项目统一的规则和标准实时的状态监控3. 自动化运维自动化的应用部署和管理实时的健康检查和告警简化的运维操作4. 教育和学习直观的系统架构演示便于理解和学习三层架构支持实验和实践后续扩展方向云端集成支持连接云端资源组建局部网络AI增强引入人工智能技术提升规则生成和应用优化能力分布式部署支持多节点部署和负载均衡安全性增强完善的权限管理和安全机制监控和分析更丰富的监控指标和分析工具总结多子项目协同中枢系统通过创新的双架构融合设计实现了信息系统架构与指令集架构的深度整合。系统采用三层架构体系支持动态重构的备忘录仪表盘通过物理层规约将映射固化在本地数据中心。这种设计不仅提高了系统的灵活性和可扩展性也为多项目协同和自动化运维提供了强大的支持。项目的成功实现验证了双架构融合设计的可行性和有效性为未来的系统设计和开发提供了新的思路和方向。本文要点-任务在真正进入开发之前我还是想确认一些必须确认的东西。看一下 我对前面讨论的“信息系统架构语言文档 抽象层 与指令集架构程序文件 逻辑层的双架构融合设计- 物理层规约库文本 ”的相关内容的重新思考请您理解我的意图开始讨论最终检查之前给出的最后设计是否需要修改。--这就是今天的任务。1、忠实原文的备忘录仪表盘 程序改造 原有 备忘录仪表盘 程序 为 “整体设计”的 多子项目 协同 的设计 。此时备忘录 则相当于 它们的协同“看板” 执行历史 结果 和当前进度-位置 和“指挥棒”控制节拍-前进速度。这个改造 包括 备忘录 设计层内部 和 实现层对外接口上的改造 根据备忘录 动态整合 所有可以利用的资源内部多任务协同 和外部 多资源协作 。这就需要 在 备忘录同时 做笔记--请求信息系统架构让自己的标引符号变成附加其上的变量对象约束 和 记日志--要求指令集架构让自己的索引文字成为 附属其下的常量 目标绑定即备忘录 的结构需要支持重构而不是现在的固定结构 并建立一个灵活的模块化框架 来支持这个重构将允许的变化限制在指令集架构的执行能力范围以内同时 仪表盘的 行为也需要重组 而不是现在的机械运动遍历自己的三个篮子而是要 能随人类行为而动重定位重配置重命名等 。这就需要 一个能动态更新的 动态组织社区 来支撑这个仪表盘通过设计不同的量词 来控制不同的变量并将受控变量和本地可用的指令 对应起来映射系统化最后就是这个映射 固化在本地数据中心最小核心实现- 通过物理层规约 。到这里 完成了本地AI数据中心 的 设计和程序。后面的问题还有 连接云端 资源 组建局部网络。但是方法相同具体函数要素和实际承担关系不同罢了。这些 考虑因素 应该体现在整体设计的整体表述中即表述内容不同维持结构-分别内容表(本地数据中心 特定AI角色 --中蕴式AI组件 仅需要本地数据原型样本选择 选择管道 数据直通访问) /真值表(局域网服务器 个性化Agent呈现样式--外挂式AI系统 请求先验判断模型标本评估 判断堆栈 信息实时处理)/ 和词汇表(云端服务 普遍性Transformer交换格式--内嵌式AI芯片 要求先天知识模具副本决策 决策快照知识动态更新不变 。这可能需要 重提之前讨论给出的两段文字做了必要的补充“知识图谱的RDFOWL GUI三者需要的 正是由 我们之前 Designer的 项目目录中第三组 原型开发 的.txt下 commandsrequests和requirements 。---您还能“记”起来吗 它们三个 正是 整体设计 原型脚本模板中 变量占位符 由 三个逻辑专项描述程序的 对应术语的确定运用方式 约简/映射/扩展 --文档--对应 实现-细节/设计-概要/需求 库。这些内容几乎 贯穿了了我们的 所有讨论将讨论过程中的阶段性结论 串起来了”“请按照讨论确定的 内容完整给出“整体设计”的设计作为后续开发的完整项目文件开发项目程序文件creator面向目标聚变-核聚变式工业实践革新 η介子自导自演的介子型DNA计算机 核糖体“I”主观现实 目标和实现意欲库 基于的 笔记本工件库域防腐层( 直接宣告和宣称的 --自述格式声明样式)ANIs DEVELOPMENT.md模板 搭建开发环境 builder面向结果(突现-基因突变式科学创新 π介子主导的中子型电子计算机 叶绿体“green”主体间和间观现实) 目的和设计意图程序中使用的 面向对象的日志表工具箱 宏隐藏层事先被隐藏的--叙述命令CLIs start.sh 模板 配套开发指南checker从“捡”然后“拣”最后“检” 面向过程渐显-循序渐进式现象维新 k介子主演的质子型量子计算机 线粒体“tree”客观现实 目光和自由意志 文档中提及的文字 到 网络备忘录仪表盘 微隔离层预先被悬置的--表述格式解释样式GUIs --readme.txt模板 。三者分别是整体设计 designer标准模板 index.html 中的三种 变量占位符 的 逻辑描述模板”--注意在第二段文字中每一段 由‘--’引导的文字 **两侧三对文字每一对分别有确定的固定选项前 格式- 文档编辑图式概念创作 专文后 样式 -程序编程模式逻辑运作 专项。而前面是 具有固定结构的不同固定方法可以有很多版本演绎但每一次结构完整性要求相同 且原始版本只有一个。这个原始版本正式 整体设计 最先要交付的基础实现。QAQ112Q112、(前述“本文要点-任务”全部文字)用户想要在开发前确认多子项目协同中枢的最终设计是否需要修改重点关注信息系统架构与指令集架构的双架构融合设计。我需要先理解用户的意图然后分析现有设计是否需要调整。任务已完成简单应用_1212 V1 创建时间08:51app_generate_app.py#!/usr/bin/env python3 # -*- coding: utf-8 -*- 应用生成HTTP服务 提供RESTful API接口来生成和管理应用 import sys import os import json from http.server import HTTPServer, BaseHTTPRequestHandler from urllib.parse import parse_qs, urlparse # 添加lib目录到路径 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from lib.generate_lib import AppGenerator class GenerateHTTPHandler(BaseHTTPRequestHandler): 应用生成HTTP处理器 def __init__(self, *args, **kwargs): 初始化处理器 # 获取正确的数据库路径 base_dir os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) db_path os.path.join(base_dir, data, app_db.json) self.generator AppGenerator(db_path) super().__init__(*args, **kwargs) def do_GET(self): 处理GET请求 parsed_url urlparse(self.path) path parsed_url.path query_params parse_qs(parsed_url.query) try: if path /api/apps: # 查询应用列表 filters {} # 处理查询参数 if type in query_params: filters[type] query_params[type][0] if status in query_params: filters[status] query_params[status][0] if keyword in query_params: filters[keyword] query_params[keyword][0] apps self.generator.get_apps(filters) self._send_response(200, {apps: apps, total: len(apps)}) elif path.startswith(/api/apps/): # 获取单个应用 app_id path.split(/)[-1] app self.generator.get_app(app_id) if app: self._send_response(200, {app: app}) else: self._send_response(404, {error: 应用不存在}) elif path /api/app-types: # 获取支持的应用类型 app_types self.generator.get_app_types() self._send_response(200, {app_types: app_types}) elif path /api/stats: # 获取统计信息 stats self.generator.get_app_stats() self._send_response(200, {stats: stats}) elif path /: # 提供前端页面 self._serve_static_file(page/app_generate.html) else: # 尝试提供静态文件 self._serve_static_file(path.lstrip(/)) except Exception as e: self._send_response(500, {error: f服务器内部错误: {str(e)}}) def do_POST(self): 处理POST请求 parsed_url urlparse(self.path) path parsed_url.path try: content_length int(self.headers.get(Content-Length, 0)) body self.rfile.read(content_length).decode(utf-8) data json.loads(body) if body else {} if path /api/apps: # 生成新应用 result self.generator.generate_app(data) if result[success]: self._send_response(201, result) else: self._send_response(400, result) elif path.startswith(/api/apps/) and /export in path: # 导出应用 app_id path.split(/)[-2] export_path data.get(export_path, /tmp) result self.generator.export_app(app_id, export_path) if result[success]: self._send_response(200, result) else: self._send_response(400, result) else: self._send_response(404, {error: 接口不存在}) except json.JSONDecodeError: self._send_response(400, {error: 无效的JSON格式}) except Exception as e: self._send_response(500, {error: f服务器内部错误: {str(e)}}) def do_DELETE(self): 处理DELETE请求 parsed_url urlparse(self.path) path parsed_url.path try: if path.startswith(/api/apps/): # 删除应用 app_id path.split(/)[-1] result self.generator.delete_app(app_id) if result[success]: self._send_response(200, result) else: self._send_response(400, result) else: self._send_response(404, {error: 接口不存在}) except Exception as e: self._send_response(500, {error: f服务器内部错误: {str(e)}}) def _send_response(self, status_code: int, data: dict): 发送JSON响应 self.send_response(status_code) self.send_header(Content-Type, application/json) self.send_header(Access-Control-Allow-Origin, *) self.send_header(Access-Control-Allow-Methods, GET, POST, PUT, DELETE, OPTIONS) self.send_header(Access-Control-Allow-Headers, Content-Type, Authorization) self.end_headers() response_body json.dumps(data, ensure_asciiFalse).encode(utf-8) self.wfile.write(response_body) def _serve_static_file(self, file_path: str): 提供静态文件 base_dir os.path.dirname(os.path.dirname(os.path.abspath(__file__))) full_path os.path.join(base_dir, file_path) # 安全检查防止目录遍历 if not full_path.startswith(base_dir): self._send_response(403, {error: 访问被拒绝}) return # 检查文件是否存在 if not os.path.exists(full_path) or not os.path.isfile(full_path): self._send_response(404, {error: 文件不存在}) return # 确定MIME类型 mime_type self._get_mime_type(full_path) # 发送文件 self.send_response(200) self.send_header(Content-Type, mime_type) self.send_header(Access-Control-Allow-Origin, *) self.end_headers() with open(full_path, rb) as f: self.wfile.write(f.read()) def _get_mime_type(self, file_path: str) - str: 获取文件的MIME类型 extensions { .html: text/html, .css: text/css, .js: application/javascript, .json: application/json, .png: image/png, .jpg: image/jpeg, .jpeg: image/jpeg, .gif: image/gif, .svg: image/svgxml, .txt: text/plain } _, ext os.path.splitext(file_path.lower()) return extensions.get(ext, application/octet-stream) def do_OPTIONS(self): 处理OPTIONS请求 self.send_response(200) self.send_header(Access-Control-Allow-Origin, *) self.send_header(Access-Control-Allow-Methods, GET, POST, PUT, DELETE, OPTIONS) self.send_header(Access-Control-Allow-Headers, Content-Type, Authorization) self.end_headers() def run_server(port: int 8002): 启动HTTP服务器 server_address (, port) httpd HTTPServer(server_address, GenerateHTTPHandler) print(f应用生成服务启动成功监听端口: {port}) print(f访问地址: http://localhost:{port}) print(fAPI文档: http://localhost:{port}/api) try: httpd.serve_forever() except KeyboardInterrupt: print(\n服务器正在关闭...) httpd.server_close() print(服务器已关闭) if __name__ __main__: # 启动服务器 run_servergenerate_lib.py#!/usr/bin/env python3 # -*- coding: utf-8 -*- 应用生成核心库 - 基于规则自动生成可运行应用 支持动态代码生成、部署和管理 import json import os import uuid import subprocess import shutil import tempfile from datetime import datetime from typing import Dict, List, Any, Optional import sys class AppGenerator: 应用生成器 - 基于规则自动生成可运行应用 def __init__(self, db_path: str ../data/app_db.json): 初始化应用生成器 self.db_path os.path.abspath(os.path.join(os.path.dirname(__file__), db_path)) self.apps self._load_apps() self.base_port 8010 self.templates self._get_templates() def _load_apps(self) - Dict[str, Any]: 加载应用数据库 try: if os.path.exists(self.db_path): with open(self.db_path, r, encodingutf-8) as f: return json.load(f) return {apps: {}, templates: {}, version: 1.0} except Exception as e: print(f加载应用数据库失败: {e}) return {apps: {}, templates: {}, version: 1.0} def _save_apps(self): 保存应用数据库 try: os.makedirs(os.path.dirname(self.db_path), exist_okTrue) with open(self.db_path, w, encodingutf-8) as f: json.dump(self.apps, f, ensure_asciiFalse, indent2) except Exception as e: print(f保存应用数据库失败: {e}) raise def _get_templates(self) - Dict[str, str]: 获取应用模板 return { simple_http: #!/usr/bin/env python3 # -*- coding: utf-8 -*- \\\ 自动生成的HTTP应用 应用ID: {{app_id}} 生成时间: {{generated_at}} \\\ import http.server import socketserver import json import os from datetime import datetime class AppHTTPHandler(http.server.BaseHTTPRequestHandler): def do_GET(self): if self.path /: self.send_response(200) self.send_header(Content-Type, text/html) self.end_headers() html !DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title{{app_name}}/title style body { font-family: Arial, sans-serif; display: flex; justify-content: center; align-items: center; min-height: 100vh; margin: 0; background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); color: #333; } .container { text-align: center; background: white; padding: 40px; border-radius: 15px; box-shadow: 0 10px 30px rgba(0,0,0,0.2); } h1 { color: #4a5568; margin-bottom: 20px; } p { color: #718096; font-size: 18px; } .info { margin-top: 30px; padding: 20px; background: #f7fafc; border-radius: 8px; text-align: left; } /style /head body div classcontainer h1 {{app_name}}/h1 p这是一个自动生成的HTTP应用/p div classinfo pstrong应用ID:/strong {{app_id}}/p pstrong生成时间:/strong {{generated_at}}/p pstrong端口:/strong {{port}}/p pstrong描述:/strong {{app_description}}/p /div /div /body /html self.wfile.write(html.encode(utf-8)) elif self.path /api/status: self.send_response(200) self.send_header(Content-Type, application/json) self.end_headers() status { status: running, app_id: {{app_id}}, app_name: {{app_name}}, timestamp: datetime.now().isoformat() } self.wfile.write(json.dumps(status).encode(utf-8)) else: self.send_response(404) self.send_header(Content-Type, application/json) self.end_headers() error {error: Not Found} self.wfile.write(json.dumps(error).encode(utf-8)) def do_POST(self): if self.path /api/data: content_length int(self.headers.get(Content-Length, 0)) body self.rfile.read(content_length).decode(utf-8) data json.loads(body) if body else {} self.send_response(200) self.send_header(Content-Type, application/json) self.end_headers() response { received: data, message: 数据接收成功, timestamp: datetime.now().isoformat() } self.wfile.write(json.dumps(response).encode(utf-8)) else: self.send_response(404) self.send_header(Content-Type, application/json) self.end_headers() error {error: Not Found} self.wfile.write(json.dumps(error).encode(utf-8)) def run_server(port): handler AppHTTPHandler with socketserver.TCPServer((, port), handler) as httpd: print(f应用启动成功监听端口: {port}) print(f访问地址: http://localhost:{port}) try: httpd.serve_forever() except KeyboardInterrupt: print(\n应用正在关闭...) httpd.server_close() if __name__ __main__: port {{port}} run_server(port) , data_processor: #!/usr/bin/env python3 # -*- coding: utf-8 -*- \\\ 自动生成的数据处理应用 应用ID: {{app_id}} 生成时间: {{generated_at}} \\\ import http.server import socketserver import json import os import time from datetime import datetime class DataProcessorHandler(http.server.BaseHTTPRequestHandler): def do_GET(self): if self.path /: self.send_response(200) self.send_header(Content-Type, text/html) self.end_headers() html !DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title{{app_name}}/title style body { font-family: Arial, sans-serif; background: #f0f2f5; margin: 0; padding: 20px; } .container { max-width: 800px; margin: 0 auto; background: white; padding: 30px; border-radius: 10px; box-shadow: 0 2px 10px rgba(0,0,0,0.1); } h1 { color: #1890ff; margin-bottom: 20px; } .form-group { margin-bottom: 20px; } textarea { width: 100%; height: 100px; padding: 10px; border: 1px solid #ddd; border-radius: 5px; resize: vertical; } button { background: #1890ff; color: white; border: none; padding: 10px 20px; border-radius: 5px; cursor: pointer; } button:hover { background: #40a9ff; } .result { margin-top: 20px; padding: 20px; background: #f6ffed; border: 1px solid #b7eb8f; border-radius: 5px; } /style /head body div classcontainer h1 {{app_name}}/h1 div classform-group label fordataInput输入数据 (JSON格式):/label textarea iddataInput placeholder{numbers: [1, 2, 3, 4, 5]}/textarea /div button onclickprocessData()处理数据/button div idresult classresult styledisplay: none;/div /div script function processData() { const input document.getElementById(dataInput).value; const resultDiv document.getElementById(result); fetch(/api/process, { method: POST, headers: { Content-Type: application/json }, body: input }) .then(response response.json()) .then(data { resultDiv.innerHTML h3处理结果:/h3pre JSON.stringify(data, null, 2) /pre; resultDiv.style.display block; }) .catch(error { resultDiv.innerHTML h3错误:/h3p error.message /p; resultDiv.style.display block; }); } /script /body /html self.wfile.write(html.encode(utf-8)) else: self.send_response(404) self.send_header(Content-Type, application/json) self.end_headers() error {error: Not Found} self.wfile.write(json.dumps(error).encode(utf-8)) def do_POST(self): if self.path /api/process: content_length int(self.headers.get(Content-Length, 0)) body self.rfile.read(content_length).decode(utf-8) try: data json.loads(body) # 数据处理逻辑 result self.process_data(data) self.send_response(200) self.send_header(Content-Type, application/json) self.end_headers() self.wfile.write(json.dumps(result).encode(utf-8)) except json.JSONDecodeError: self.send_response(400) self.send_header(Content-Type, application/json) self.end_headers() error {error: Invalid JSON format} self.wfile.write(json.dumps(error).encode(utf-8)) else: self.send_response(404) self.send_header(Content-Type, application/json) self.end_headers() error {error: Not Found} self.wfile.write(json.dumps(error).encode(utf-8)) def process_data(self, data): 数据处理逻辑 result { original: data, processed_at: datetime.now().isoformat(), statistics: {} } # 基本统计 if isinstance(data, dict): result[statistics][keys_count] len(data.keys()) result[statistics][value_types] {} for key, value in data.items(): value_type type(value).__name__ result[statistics][value_types][value_type] \ result[statistics][value_types].get(value_type, 0) 1 # 数字数组统计 if isinstance(value, list) and all(isinstance(x, (int, float)) for x in value): result[statistics][f{key}_stats] { count: len(value), sum: sum(value), average: sum(value) / len(value) if value else 0, min: min(value) if value else None, max: max(value) if value else None } elif isinstance(data, list): result[statistics][items_count] len(data) if all(isinstance(x, (int, float)) for x in data): result[statistics][numbers_stats] { count: len(data), sum: sum(data), average: sum(data) / len(data) if data else 0, min: min(data) if data else None, max: max(data) if data else None } return result def run_server(port): handler DataProcessorHandler with socketserver.TCPServer((, port), handler) as httpd: print(f数据处理应用启动成功监听端口: {port}) print(f访问地址: http://localhost:{port}) try: httpd.serve_forever() except KeyboardInterrupt: print(\n应用正在关闭...) httpd.server_close() if __name__ __main__: port {{port}} run_server(port) } def generate_app(self, app_config: Dict[str, Any]) - Dict[str, Any]: 生成应用 基于规则和配置自动生成可运行的应用 try: # 验证配置 validated_config self._validate_app_config(app_config) # 生成应用ID app_id str(uuid.uuid4()) # 分配端口 port self._allocate_port() # 生成应用代码 app_code self._generate_app_code(app_id, validated_config, port) # 创建应用目录 app_dir self._create_app_directory(app_id) # 保存应用文件 app_file os.path.join(app_dir, fapp_{app_id}.py) with open(app_file, w, encodingutf-8) as f: f.write(app_code) # 设置可执行权限 os.chmod(app_file, 0o755) # 创建启动脚本 start_script os.path.join(app_dir, start.sh) with open(start_script, w, encodingutf-8) as f: f.write(f#!/bin/bash python3 app_{app_id}.py ) os.chmod(start_script, 0o755) # 创建停止脚本 stop_script os.path.join(app_dir, stop.sh) with open(stop_script, w, encodingutf-8) as f: f.write(f#!/bin/bash pkill -f python3 app_{app_id}.py ) os.chmod(stop_script, 0o755) # 保存应用信息 app_info { id: app_id, name: validated_config[name], description: validated_config[description], type: validated_config[type], port: port, directory: app_dir, generated_at: datetime.now().isoformat(), status: created, config: validated_config } self.apps[apps][app_id] app_info self._save_apps() return { success: True, app_id: app_id, port: port, message: 应用生成成功 } except Exception as e: return {success: False, message: f应用生成失败: {str(e)}} def _validate_app_config(self, app_config: Dict[str, Any]) - Dict[str, Any]: 验证应用配置 required_fields [name, description, type] for field in required_fields: if field not in app_config: raise ValueError(f缺少必需字段: {field}) # 验证应用类型 valid_types list(self.templates.keys()) if app_config[type] not in valid_types: raise ValueError(f无效的应用类型支持的类型: {, .join(valid_types)}) return app_config def _allocate_port(self) - int: 分配端口 used_ports [app[port] for app in self.apps[apps].values()] port self.base_port while port in used_ports: port 1 return port def _generate_app_code(self, app_id: str, config: Dict[str, Any], port: int) - str: 生成应用代码 template self.templates.get(config[type]) if not template: raise ValueError(f未知的应用类型: {config[type]}) # 替换模板变量 code template.replace({{app_id}}, app_id) code code.replace({{app_name}}, config[name]) code code.replace({{app_description}}, config[description]) code code.replace({{port}}, str(port)) code code.replace({{generated_at}}, datetime.now().isoformat()) return code def _create_app_directory(self, app_id: str) - str: 创建应用目录 base_dir os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) app_dir os.path.join(base_dir, generated_apps, app_id) os.makedirs(app_dir, exist_okTrue) return app_dir def get_apps(self, filters: Optional[Dict[str, Any]] None) - List[Dict[str, Any]]: 获取应用列表 try: result [] for app_id, app in self.apps[apps].items(): # 应用过滤条件 if filters: match True if type in filters and app.get(type) ! filters[type]: match False if status in filters and app.get(status) ! filters[status]: match False if keyword in filters: keyword filters[keyword].lower() if not (keyword in app.get(name, ).lower() or keyword in app.get(description, ).lower()): match False if not match: continue result.append(app) # 按生成时间排序 result.sort(keylambda x: x.get(generated_at, ), reverseTrue) return result except Exception as e: print(f获取应用列表失败: {e}) return [] def get_app(self, app_id: str) - Optional[Dict[str, Any]]: 获取单个应用信息 return self.apps[apps].get(app_id) def delete_app(self, app_id: str) - Dict[str, Any]: 删除应用 try: if app_id not in self.apps[apps]: return {success: False, message: 应用不存在 app self.apps[apps][app_id] # 删除应用目录 if os.path.exists(app[directory]): shutil.rmtree(app[directory]) # 删除应用信息 del self.apps[apps][app_id] self._save_apps() return {success: True, message: 应用删除成功} except Exception as e: return {success: False, message: f应用删除失败: {str(e)}} def get_app_types(self) - List[str]: 获取支持的应用类型 return list(self.templates.keys()) def get_app_stats(self) - Dict[str, Any]: 获取应用统计信息 total_apps len(self.apps[apps]) # 按类型统计 type_count {} status_count {} for app in self.apps[apps].values(): app_type app.get(type, unknown) status app.get(status, unknown) type_count[app_type] type_count.get(app_type, 0) 1 status_count[status] status_count.get(status, 0) 1 return { total_apps: total_apps, type_count: type_count, status_count: status_count, last_updated: datetime.now().isoformat() } def export_app(self, app_id: str, export_path: str) - Dict[str, Any]: 导出应用 try: if app_id not in self.apps[apps]: return {success: False, message: 应用不存在 app self.apps[apps][app_id] # 创建导出目录 os.makedirs(export_path, exist_okTrue) # 复制应用文件 app_files os.listdir(app[directory]) for file in app_files: src_path os.path.join(app[directory], file) dst_path os.path.join(export_path, file) shutil.copy2(src_path, dst_path) # 创建应用配置文件 config_file os.path.join(export_path, app_config.json) with open(config_file, w, encodingutf-8) as f: json.dump(app, f, ensure_asciiFalse, indent2) return {success: True, message: 应用导出成功} except Exception as e: return {success: False, message: f应用导出失败: {str(e)}} # 示例使用 if __name__ __main__: # 创建应用生成器实例 generator AppGenerator() # 生成示例应用 app_config { name: 测试应用, description: 这是一个测试应用, type: simple_http } result generator.generate_app(app_config) print(f应用生成结果: {result}) # 获取应用列表 apps generator.get_apps() print(f应用列表: {len(apps)} 个应用) # 获取统计信息 stats generator.get_app_stats() print(f应用统计: {statsapp_generate.html!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title应用生成中心 - 多子项目协同中枢/title style * { margin: 0; padding: 0; box-sizing: border-box; } body { font-family: -apple-system, BlinkMacSystemFont, Segoe UI, Roboto, Helvetica Neue, Arial, sans-serif; background: linear-gradient(135deg, #f093fb 0%, #f5576c 100%); min-height: 100vh; color: #333; } .container { max-width: 1200px; margin: 0 auto; padding: 20px; } .header { background: rgba(255, 255, 255, 0.95); backdrop-filter: blur(10px); border-radius: 15px; padding: 30px; margin-bottom: 30px; box-shadow: 0 8px 32px rgba(0, 0, 0, 0.1); text-align: center; } .header h1 { font-size: 2.5em; color: #4a5568; margin-bottom: 10px; font-weight: 700; } .header p { color: #718096; font-size: 1.1em; } .main-content { display: grid; grid-template-columns: 1fr 1fr; gap: 30px; margin-bottom: 30px; } .card { background: rgba(255, 255, 255, 0.95); backdrop-filter: blur(10px); border-radius: 15px; padding: 30px; box-shadow: 0 8px 32px rgba(0, 0, 0, 0.1); transition: transform 0.3s ease, box-shadow 0.3s ease; } .card:hover { transform: translateY(-5px); box-shadow: 0 12px 40px rgba(0, 0, 0, 0.15); } .card h2 { color: #2d3748; margin-bottom: 20px; font-size: 1.5em; border-bottom: 2px solid #e2e8f0; padding-bottom: 10px; } .form-group { margin-bottom: 20px; } .form-group label { display: block; margin-bottom: 8px; font-weight: 600; color: #4a5568; } .form-group input, .form-group select, .form-group textarea { width: 100%; padding: 12px 16px; border: 2px solid #e2e8f0; border-radius: 8px; font-size: 14px; transition: border-color 0.3s ease, box-shadow 0.3s ease; } .form-group input:focus, .form-group select:focus, .form-group textarea:focus { outline: none; border-color: #f093fb; box-shadow: 0 0 0 3px rgba(240, 147, 251, 0.1); } .form-group textarea { resize: vertical; min-height: 100px; } .btn { background: linear-gradient(135deg, #f093fb 0%, #f5576c 100%); color: white; border: none; padding: 12px 24px; border-radius: 8px; font-size: 14px; font-weight: 600; cursor: pointer; transition: transform 0.3s ease, box-shadow 0.3s ease; margin-right: 10px; margin-bottom: 10px; } .btn:hover { transform: translateY(-2px); box-shadow: 0 4px 12px rgba(240, 147, 251, 0.3); } .btn-secondary { background: linear-gradient(135deg, #a0aec0 0%, #718096 100%); } .btn-danger { background: linear-gradient(135deg, #fc8181 0%, #f56565 100%); } .btn-success { background: linear-gradient(135deg, #48bb78 0%, #38a169 100%); } .btn-info { background: linear-gradient(135deg, #4299e1 0%, #3182ce 100%); } .app-list { max-height: 500px; overflow-y: auto; } .app-item { background: #f7fafc; border-radius: 8px; padding: 20px; margin-bottom: 15px; border-left: 4px solid #f093fb; transition: transform 0.2s ease, box-shadow 0.2s ease; } .app-item:hover { transform: translateX(5px); box-shadow: 0 4px 12px rgba(0, 0, 0, 0.1); } .app-item h3 { color: #2d3748; margin-bottom: 8px; font-size: 1.2em; display: flex; align-items: center; gap: 8px; } .app-item p { color: #718096; margin-bottom: 15px; font-size: 0.9em; line-height: 1.5; } .app-meta { display: flex; flex-wrap: wrap; gap: 15px; font-size: 0.8em; color: #a0aec0; margin-bottom: 15px; } .app-meta span { background: #edf2f7; padding: 4px 8px; border-radius: 4px; } .app-status { display: inline-block; padding: 4px 12px; border-radius: 20px; font-size: 0.8em; font-weight: 600; text-transform: uppercase; } .app-status.created { background: #e2e8f0; color: #4a5568; } .app-status.running { background: #c6f6d5; color: #22543d; } .app-status.stopped { background: #fed7d7; color: #742a2a; } .app-actions { display: flex; gap: 10px; flex-wrap: wrap; } .app-actions .btn { padding: 6px 12px; font-size: 12px; } .filters { display: flex; gap: 15px; margin-bottom: 20px; flex-wrap: wrap; } .filters select, .filters input { padding: 8px 12px; border: 2px solid #e2e8f0; border-radius: 6px; font-size: 14px; } .stats-grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); gap: 20px; margin-bottom: 30px; } .stat-card { background: rgba(255, 255, 255, 0.95); backdrop-filter: blur(10px); border-radius: 15px; padding: 25px; text-align: center; box-shadow: 0 8px 32px rgba(0, 0, 0, 0.1); transition: transform 0.3s ease; } .stat-card:hover { transform: translateY(-5px); } .stat-number { font-size: 2.5em; font-weight: 700; color: #f093fb; margin-bottom: 10px; } .stat-label { color: #718096; font-size: 1.1em; } .notification { position: fixed; top: 20px; right: 20px; padding: 15px 20px; border-radius: 8px; color: white; font-weight: 600; z-index: 1000; transform: translateX(400px); transition: transform 0.3s ease; } .notification.show { transform: translateX(0); } .notification.success { background: #48bb78; } .notification.error { background: #f56565; } .notification.info { background: #4299e1; } .app-type-info { background: #edf2f7; padding: 15px; border-radius: 8px; margin-bottom: 20px; border-left: 4px solid #4299e1; } .app-type-info h4 { color: #2d3748; margin-bottom: 8px; } .app-type-info p { color: #718096; font-size: 0.9em; margin: 0; } .loading { display: inline-block; width: 20px; height: 20px; border: 3px solid #f3f3f3; border-top: 3px solid #f093fb; border-radius: 50%; animation: spin 1s linear infinite; } keyframes spin { 0% { transform: rotate(0deg); } 100% { transform: rotate(360deg); } } .empty-state { text-align: center; padding: 40px 20px; color: #a0aec0; } .empty-state i { font-size: 4em; margin-bottom: 20px; opacity: 0.5; } .icon { width: 24px; height: 24px; display: inline-block; vertical-align: middle; } media (max-width: 768px) { .main-content { grid-template-columns: 1fr; } .header h1 { font-size: 2em; } .filters { flex-direction: column; } .stats-grid { grid-template-columns: 1fr; } } /style /head body div classcontainer !-- 头部 -- div classheader h1 应用生成中心/h1 p基于规则自动生成可运行应用支持动态代码生成和部署/p /div !-- 统计信息 -- div classstats-grid div classstat-card div classstat-number idtotalApps0/div div classstat-label总应用数/div /div div classstat-card div classstat-number idrunningApps0/div div classstat-label运行中应用/div /div div classstat-card div classstat-number idappTypes0/div div classstat-label应用类型/div /div /div !-- 主要内容 -- div classmain-content !-- 应用生成 -- div classcard h2✨ 生成新应用/h2 div classapp-type-info idappTypeInfo h4选择应用类型/h4 p不同的应用类型提供不同的功能和特性请根据需求选择合适的类型。/p /div form idappForm div classform-group label forappName应用名称 */label input typetext idappName required placeholder输入应用名称 /div div classform-group label forappDescription应用描述 */label textarea idappDescription required placeholder输入应用描述/textarea /div div classform-group label forappType应用类型 */label select idappType required onchangeupdateAppTypeInfo() option value选择应用类型/option /select /div div stylemargin-top: 20px; button typesubmit classbtn span idgenerateBtnText生成应用/span span idloadingIndicator classloading styledisplay: none;/span /button button typebutton classbtn btn-secondary onclickclearForm() 清空表单 /button /div /form /div !-- 应用列表 -- div classcard h2 应用列表/h2 !-- 过滤器 -- div classfilters select idtypeFilter onchangeloadApps() option value所有类型/option /select select idstatusFilter onchangeloadApps() option value所有状态/option option valuecreated已创建/option option valuerunning运行中/option option valuestopped已停止/option /select input typetext idkeywordFilter placeholder搜索关键词 oninputloadApps() /div div classapp-list idappList div classempty-state div/div p暂无应用请生成新应用/p /div /div /div /div /div !-- 通知 -- div classnotification idnotification/div script // 应用类型信息 const appTypeDescriptions { simple_http: { title: 简单HTTP应用, description: 基础的HTTP服务器应用提供Web界面和API接口适合快速原型开发和简单的Web服务。, features: [Web界面, RESTful API, 静态文件服务, 简单数据处理] }, data_processor: { title: 数据处理应用, description: 专门用于数据处理和分析的应用支持JSON数据的接收、处理和响应适合数据转换和统计分析。, features: [数据接收, 实时处理, 统计分析, 可视化结果] } }; // 初始化 document.addEventListener(DOMContentLoaded, function() { loadAppTypes(); loadApps(); loadStats(); setupForm(); }); // 设置表单 function setupForm() { document.getElementById(appForm).addEventListener(submit, function(e) { e.preventDefault(); generateApp(); }); } // 加载应用类型 function loadAppTypes() { fetch(/api/app-types) .then(response response.json()) .then(data { const appTypeSelect document.getElementById(appType); const typeFilter document.getElementById(typeFilter); data.app_types.forEach(app_type { const option1 document.createElement(option); option1.value app_type; option1.textContent appTypeDescriptions[app_type]?.title || app_type; appTypeSelect.appendChild(option1); const option2 document.createElement(option); option2.value app_type; option2.textContent appTypeDescriptions[app_type]?.title || app_type; typeFilter.appendChild(option2); }); }) .catch(error { showNotification(加载应用类型失败, error); }); } // 更新应用类型信息 function updateAppTypeInfo() { const appType document.getElementById(appType).value; const infoDiv document.getElementById(appTypeInfo); if (appType appTypeDescriptions[appType]) { const info appTypeDescriptions[appType]; infoDiv.innerHTML h4${info.title}/h4 p${info.description}/p div stylemargin-top: 10px; strong特性/strong span stylecolor: #4299e1; font-size: 0.8em; ${info.features.join( • )} /span /div ; } else { infoDiv.innerHTML h4选择应用类型/h4 p不同的应用类型提供不同的功能和特性请根据需求选择合适的类型。/p ; } } // 加载应用列表 function loadApps() { const type document.getElementById(typeFilter).value; const status document.getElementById(statusFilter).value; const keyword document.getElementById(keywordFilter).value; let url /api/apps?; if (type) url type${type}; if (status) url status${status}; if (keyword) url keyword${keyword}; fetch(url) .then(response response.json()) .then(data { renderApps(data.apps); }) .catch(error { showNotification(加载应用列表失败, error); }); } // 渲染应用列表 function renderApps(apps) { const appList document.getElementById(appList); if (apps.length 0) { appList.innerHTML div classempty-state div/div p没有找到符合条件的应用/p /div ; return; } appList.innerHTML apps.map(app div classapp-item h3 ${getAppIcon(app.type)} ${app.name} /h3 p${app.description}/p div classapp-meta span类型: ${appTypeDescriptions[app.type]?.title || app.type}/span span端口: ${app.port}/span span生成时间: ${formatTime(app.generated_at)}/span /div div stylemargin-bottom: 15px; 状态: span classapp-status ${app.status}${getStatusText(app.status)}/span /div div classapp-actions button classbtn btn-info onclickviewApp(${app.id}, ${app.port}) 查看 /button button classbtn btn-success onclickstartApp(${app.id}) 启动 /button button classbtn btn-secondary onclickstopApp(${app.id}) 停止 /button button classbtn btn-danger onclickdeleteApp(${app.id}) 删除 /button /div /div ).join(); } // 获取应用图标 function getAppIcon(appType) { const icons { simple_http: , data_processor: }; return icons[appType] || ; } // 获取状态文本 function getStatusText(status) { const statusMap { created: 已创建, running: 运行中, stopped: 已停止 }; return statusMap[status] || status; } // 格式化时间 function formatTime(timeString) { const date new Date(timeString); return date.toLocaleString(zh-CN, { year: numeric, month: 2-digit, day: 2-digit, hour: 2-digit, minute: 2-digit }); } // 加载统计信息 function loadStats() { fetch(/api/stats) .then(response response.json()) .then(data { document.getElementById(totalApps).textContent data.stats.total_apps; // 计算运行中应用数 const runningCount data.stats.status_count?.running || 0; document.getElementById(runningApps).textContent runningCount; // 应用类型数量 const typeCount Object.keys(data.stats.type_count || {}).length; document.getElementById(appTypes).textContent typeCount; }) .catch(error { console.error(加载统计信息失败:, error); }); } // 生成应用 function generateApp() { const appData { name: document.getElementById(appName).value, description: document.getElementById(appDescription).value, type: document.getElementById(appType).value }; // 显示加载状态 document.getElementById(generateBtnText).textContent 生成中...; document.getElementById(loadingIndicator).style.display inline-block; fetch(/api/apps, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify(appData) }) .then(response response.json()) .then(data { // 恢复按钮状态 document.getElementById(generateBtnText).textContent 生成应用; document.getElementById(loadingIndicator).style.display none; if (data.success) { showNotification(应用生成成功, success); clearForm(); loadApps(); loadStats(); // 自动跳转到应用运行页面 setTimeout(() { window.open(http://localhost:8003, _blank); }, 1000); } else { showNotification(data.message || 生成失败, error); } }) .catch(error { // 恢复按钮状态 document.getElementById(generateBtnText).textContent 生成应用; document.getElementById(loadingIndicator).style.display none; showNotification(生成应用失败, error); }); } // 查看应用 function viewApp(appId, port) { window.open(http://localhost:${port}, _blank); } // 启动应用 function startApp(appId) { // 这里需要调用应用运行层的API // 暂时显示提示信息 showNotification(请在应用运行中心启动应用, info); window.open(http://localhost:8003, _blank); } // 停止应用 function stopApp(appId) { // 这里需要调用应用运行层的API // 暂时显示提示信息 showNotification(请在应用运行中心停止应用, info); window.open(http://localhost:8003, _blank); } // 删除应用 function deleteApp(appId) { if (confirm(确定要删除这个应用吗此操作不可恢复。)) { fetch(/api/apps/${appId}, { method: DELETE }) .then(response response.json()) .then(data { if (data.success) { showNotification(应用删除成功, success); loadApps(); loadStats(); } else { showNotification(data.message || 删除失败, error); } }) .catch(error { showNotification(删除应用失败, error); }); } } // 清空表单 function clearForm() { document.getElementById(appForm).reset(); updateAppTypeInfo(); } // 显示通知 function showNotification(message, type info) { const notification document.getElementById(notification); notification.textContent message; notification.className notification ${type} show; setTimeout(() { notification.classList.remove(show); }, 3000); } // 自动刷新 setInterval(() { loadStats(); }, 30000); // 每30秒刷新一次统计信息 /script /body /htmlrule_define_app.py#!/usr/bin/env python3 # -*- coding: utf-8 -*- 规则定义HTTP服务 提供RESTful API接口来管理规则 import sys import os import json from http.server import HTTPServer, BaseHTTPRequestHandler from urllib.parse import parse_qs, urlparse # 添加lib目录到路径 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from lib.rule_lib import RuleManager class RuleHTTPHandler(BaseHTTPRequestHandler): 规则HTTP处理器 def __init__(self, *args, **kwargs): 初始化处理器 # 获取正确的数据库路径 base_dir os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) db_path os.path.join(base_dir, data, rule_db.json) self.rule_manager RuleManager(db_path) super().__init__(*args, **kwargs) def do_GET(self): 处理GET请求 parsed_url urlparse(self.path) path parsed_url.path query_params parse_qs(parsed_url.query) try: if path /api/rules: # 查询规则列表 filters {} # 处理查询参数 if category in query_params: filters[category] query_params[category][0] if active in query_params: filters[active] query_params[active][0].lower() true if min_priority in query_params: filters[min_priority] int(query_params[min_priority][0]) if keyword in query_params: filters[keyword] query_params[keyword][0] rules self.rule_manager.query_rules(filters) self._send_response(200, {rules: rules, total: len(rules)}) elif path.startswith(/api/rules/): # 获取单个规则 rule_id path.split(/)[-1] rules self.rule_manager.query_rules({rule_id: rule_id}) if rules: self._send_response(200, {rule: rules[0]}) else: self._send_response(404, {error: 规则不存在}) elif path /api/categories: # 获取分类列表 categories self.rule_manager.get_categories() self._send_response(200, {categories: categories}) elif path /api/stats: # 获取统计信息 stats self.rule_manager.get_rule_stats() self._send_response(200, {stats: stats}) elif path /: # 提供前端页面 self._serve_static_file(page/rule_define.html) else: # 尝试提供静态文件 self._serve_static_file(path.lstrip(/)) except Exception as e: self._send_response(500, {error: f服务器内部错误: {str(e)}}) def do_POST(self): 处理POST请求 parsed_url urlparse(self.path) path parsed_url.path try: content_length int(self.headers.get(Content-Length, 0)) body self.rfile.read(content_length).decode(utf-8) data json.loads(body) if body else {} if path /api/rules: # 注册新规则 result self.rule_manager.register_rule(data) if result[success]: self._send_response(201, result) else: self._send_response(400, result) elif path.startswith(/api/rules/) and /validate in path: # 验证规则条件 rule_id path.split(/)[-2] result { valid: self.rule_manager.validate_rule_conditions(rule_id, data) } self._send_response(200, result) else: self._send_response(404, {error: 接口不存在}) except json.JSONDecodeError: self._send_response(400, {error: 无效的JSON格式}) except Exception as e: self._send_response(500, {error: f服务器内部错误: {str(e)}}) def do_PUT(self): 处理PUT请求 parsed_url urlparse(self.path) path parsed_url.path try: content_length int(self.headers.get(Content-Length, 0)) body self.rfile.read(content_length).decode(utf-8) data json.loads(body) if body else {} if path.startswith(/api/rules/): # 更新规则 rule_id path.split(/)[-1] result self.rule_manager.update_rule(rule_id, data) if result[success]: self._send_response(200, result) else: self._send_response(400, result) else: self._send_response(404, {error: 接口不存在}) except json.JSONDecodeError: self._send_response(400, {error: 无效的JSON格式}) except Exception as e: self._send_response(500, {error: f服务器内部错误: {str(e)}}) def do_DELETE(self): 处理DELETE请求 parsed_url urlparse(self.path) path parsed_url.path try: if path.startswith(/api/rules/): # 删除规则 rule_id path.split(/)[-1] result self.rule_manager.delete_rule(rule_id) if result[success]: self._send_response(200, result) else: self._send_response(400, result) else: self._send_response(404, {error: 接口不存在}) except Exception as e: self._send_response(500, {error: f服务器内部错误: {str(e)}}) def _send_response(self, status_code: int, data: dict): 发送JSON响应 self.send_response(status_code) self.send_header(Content-Type, application/json) self.send_header(Access-Control-Allow-Origin, *) self.send_header(Access-Control-Allow-Methods, GET, POST, PUT, DELETE, OPTIONS) self.send_header(Access-Control-Allow-Headers, Content-Type, Authorization) self.end_headers() response_body json.dumps(data, ensure_asciiFalse).encode(utf-8) self.wfile.write(response_body) def _serve_static_file(self, file_path: str): 提供静态文件 base_dir os.path.dirname(os.path.dirname(os.path.abspath(__file__))) full_path os.path.join(base_dir, file_path) # 安全检查防止目录遍历 if not full_path.startswith(base_dir): self._send_response(403, {error: 访问被拒绝}) return # 检查文件是否存在 if not os.path.exists(full_path) or not os.path.isfile(full_path): self._send_response(404, {error: 文件不存在}) return # 确定MIME类型 mime_type self._get_mime_type(full_path) # 发送文件 self.send_response(200) self.send_header(Content-Type, mime_type) self.send_header(Access-Control-Allow-Origin, *) self.end_headers() with open(full_path, rb) as f: self.wfile.write(f.read()) def _get_mime_type(self, file_path: str) - str: 获取文件的MIME类型 extensions { .html: text/html, .css: text/css, .js: application/javascript, .json: application/json, .png: image/png, .jpg: image/jpeg, .jpeg: image/jpeg, .gif: image/gif, .svg: image/svgxml, .txt: text/plain } _, ext os.path.splitext(file_path.lower()) return extensions.get(ext, application/octet-stream) def do_OPTIONS(self): 处理OPTIONS请求 self.send_response(200) self.send_header(Access-Control-Allow-Origin, *) self.send_header(Access-Control-Allow-Methods, GET, POST, PUT, DELETE, OPTIONS) self.send_header(Access-Control-Allow-Headers, Content-Type, Authorization) self.end_headers() def run_server(port: int 8001): 启动HTTP服务器 server_address (, port) httpd HTTPServer(server_address, RuleHTTPHandler) print(f规则定义服务启动成功监听端口: {port}) print(f访问地址: http://localhost:{port}) print(fAPI文档: http://localhost:{port}/api) try: httpd.serve_forever() except KeyboardInterrupt: print(\n服务器正在关闭...) httpd.server_close() print(服务器已关闭) if __name__ __main__: # 启动服务器 run_serverrule_lib.py#!/usr/bin/env python3 # -*- coding: utf-8 -*- 规则处理核心库 - 支持动态重构的规则管理系统 实现备忘录仪表盘的规则定义功能支持规则注册、查询和校验 import json import os import uuid from datetime import datetime from typing import Dict, List, Any, Optional class RuleManager: 规则管理器 - 支持动态重构的规则管理系统 def __init__(self, db_path: str ../data/rule_db.json): 初始化规则管理器 self.db_path os.path.abspath(os.path.join(os.path.dirname(__file__), db_path)) self.rules self._load_rules() self.schema self._get_default_schema() def _load_rules(self) - Dict[str, Any]: 加载规则数据库 try: if os.path.exists(self.db_path): with open(self.db_path, r, encodingutf-8) as f: return json.load(f) return {rules: {}, categories: {}, version: 1.0} except Exception as e: print(f加载规则数据库失败: {e}) return {rules: {}, categories: {}, version: 1.0} def _save_rules(self): 保存规则数据库 try: os.makedirs(os.path.dirname(self.db_path), exist_okTrue) with open(self.db_path, w, encodingutf-8) as f: json.dump(self.rules, f, ensure_asciiFalse, indent2) except Exception as e: print(f保存规则数据库失败: {e}) raise def _get_default_schema(self) - Dict[str, Any]: 获取默认规则架构 return { name: {type: string, required: True}, description: {type: string, required: True}, category: {type: string, required: True}, priority: {type: integer, default: 0}, active: {type: boolean, default: True}, conditions: {type: array, default: []}, actions: {type: array, default: []}, metadata: {type: object, default: {}} } def register_rule(self, rule_data: Dict[str, Any]) - Dict[str, Any]: 注册新规则 支持动态重构的规则注册包含变量约束和目标绑定 try: # 验证规则数据 validated_data self._validate_rule(rule_data) # 生成规则ID rule_id str(uuid.uuid4()) # 添加元数据 validated_data[id] rule_id validated_data[created_at] datetime.now().isoformat() validated_data[updated_at] datetime.now().isoformat() # 保存规则 self.rules[rules][rule_id] validated_data # 更新分类 category validated_data.get(category) if category: if category not in self.rules[categories]: self.rules[categories][category] [] if rule_id not in self.rules[categories][category]: self.rules[categories][category].append(rule_id) self._save_rules() return {success: True, rule_id: rule_id, message: 规则注册成功} except Exception as e: return {success: False, message: f规则注册失败: {str(e)}} def _validate_rule(self, rule_data: Dict[str, Any]) - Dict[str, Any]: 验证规则数据 validated {} # 应用默认值并验证 for field, config in self.schema.items(): if field in rule_data: value rule_data[field] # 类型验证 if config[type] string and not isinstance(value, str): raise ValueError(f字段 {field} 必须是字符串类型) elif config[type] integer and not isinstance(value, int): raise ValueError(f字段 {field} 必须是整数类型) elif config[type] boolean and not isinstance(value, bool): raise ValueError(f字段 {field} 必须是布尔类型) elif config[type] array and not isinstance(value, list): raise ValueError(f字段 {field} 必须是数组类型) elif config[type] object and not isinstance(value, dict): raise ValueError(f字段 {field} 必须是对象类型) validated[field] value elif config.get(required, False): raise ValueError(f字段 {field} 是必需的) else: validated[field] config.get(default) # 添加额外的字段 for key, value in rule_data.items(): if key not in self.schema: validated[key] value return validated def query_rules(self, filters: Optional[Dict[str, Any]] None) - List[Dict[str, Any]]: 查询规则 支持按分类、状态、优先级等条件筛选 try: result [] for rule_id, rule in self.rules[rules].items(): # 应用过滤条件 if filters: match True # 按分类过滤 if category in filters and rule.get(category) ! filters[category]: match False # 按状态过滤 if active in filters and rule.get(active) ! filters[active]: match False # 按优先级过滤 if min_priority in filters and rule.get(priority, 0) filters[min_priority]: match False # 按关键词搜索 if keyword in filters: keyword filters[keyword].lower() if not (keyword in rule.get(name, ).lower() or keyword in rule.get(description, ).lower()): match False if not match: continue result.append(rule) # 按优先级排序 result.sort(keylambda x: x.get(priority, 0), reverseTrue) return result except Exception as e: print(f查询规则失败: {e}) return [] def update_rule(self, rule_id: str, updates: Dict[str, Any]) - Dict[str, Any]: 更新规则 支持动态重构规则 try: if rule_id not in self.rules[rules]: return {success: False, message: 规则不存在 # 获取原规则 old_rule self.rules[rules][rule_id].copy() # 验证更新数据 update_data old_rule.copy() update_data.update(updates) validated_data self._validate_rule(update_data) # 更新时间戳 validated_data[updated_at] datetime.now().isoformat() # 处理分类变更 old_category old_rule.get(category) new_category validated_data.get(category) if old_category ! new_category: # 从旧分类中移除 if old_category and old_category in self.rules[categories]: self.rules[categories][old_category] [ id for id in self.rules[categories][old_category] if id ! rule_id ] # 添加到新分类 if new_category: if new_category not in self.rules[categories]: self.rules[categories][new_category] [] if rule_id not in self.rules[categories][new_category]: self.rules[categories][new_category].append(rule_id) # 保存更新 self.rules[rules][rule_id] validated_data self._save_rules() return {success: True, message: 规则更新成功} except Exception as e: return {success: False, message: f规则更新失败: {str(e)}} def delete_rule(self, rule_id: str) - Dict[str, Any]: 删除规则 try: if rule_id not in self.rules[rules]: return {success: False, message: 规则不存在 # 获取规则信息 rule self.rules[rules][rule_id] # 从分类中移除 category rule.get(category) if category and category in self.rules[categories]: self.rules[categories][category] [ id for id in self.rules[categories][category] if id ! rule_id ] # 删除规则 del self.rules[rules][rule_id] self._save_rules() return {success: True, message: 规则删除成功} except Exception as e: return {success: False, message: f规则删除失败: {str(e)}} def get_categories(self) - List[str]: 获取所有分类 return list(self.rules[categories].keys()) def validate_rule_conditions(self, rule_id: str, context: Dict[str, Any]) - bool: 验证规则条件 支持动态上下文评估 try: if rule_id not in self.rules[rules]: return False rule self.rules[rules][rule_id] conditions rule.get(conditions, []) if not conditions: return True # 简单的条件评估逻辑 for condition in conditions: field condition.get(field) operator condition.get(operator) value condition.get(value) if field not in context: return False context_value context[field] if operator eq and context_value ! value: return False elif operator neq and context_value value: return False elif operator gt and context_value value: return False elif operator lt and context_value value: return False elif operator contains and value not in context_value: return False return True except Exception: return False def get_rule_actions(self, rule_id: str) - List[Dict[str, Any]]: 获取规则动作 if rule_id in self.rules[rules]: return self.rules[rules][rule_id].get(actions, []) return [] def get_rule_stats(self) - Dict[str, Any]: 获取规则统计信息 total_rules len(self.rules[rules]) active_rules sum(1 for rule in self.rules[rules].values() if rule.get(active, True)) categories_count {cat: len(ids) for cat, ids in self.rules[categories].items()} return { total_rules: total_rules, active_rules: active_rules, categories_count: categories_count, last_updated: datetime.now().isoformat() } # 示例使用 if __name__ __main__: # 创建规则管理器实例 rule_manager RuleManager() # 注册示例规则 sample_rule { name: 备忘录更新规则, description: 当备忘录内容更新时触发的规则, category: memo, priority: 10, conditions: [ {field: content_type, operator: eq, value: text}, {field: update_type, operator: eq, value: content_change ], actions: [ {type: log, message: 备忘录内容已更新}, {type: notify, channel: dashboard} ] } result rule_manager.register_rule(sample_rule) print(f规则注册结果: {result}) # 查询规则 rules rule_manager.query_rules({category: memo}) print(f查询到的规则数量: {len(rules)}) # 获取统计信息 stats rule_manager.get_rule_stats() print(f规则统计: {statsrule_define.html!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title规则定义中心 - 多子项目协同中枢/title style * { margin: 0; padding: 0; box-sizing: border-box; } body { font-family: -apple-system, BlinkMacSystemFont, Segoe UI, Roboto, Helvetica Neue, Arial, sans-serif; background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); min-height: 100vh; color: #333; } .container { max-width: 1200px; margin: 0 auto; padding: 20px; } .header { background: rgba(255, 255, 255, 0.95); backdrop-filter: blur(10px); border-radius: 15px; padding: 30px; margin-bottom: 30px; box-shadow: 0 8px 32px rgba(0, 0, 0, 0.1); text-align: center; } .header h1 { font-size: 2.5em; color: #4a5568; margin-bottom: 10px; font-weight: 700; } .header p { color: #718096; font-size: 1.1em; } .main-content { display: grid; grid-template-columns: 1fr 1fr; gap: 30px; margin-bottom: 30px; } .card { background: rgba(255, 255, 255, 0.95); backdrop-filter: blur(10px); border-radius: 15px; padding: 30px; box-shadow: 0 8px 32px rgba(0, 0, 0, 0.1); transition: transform 0.3s ease, box-shadow 0.3s ease; } .card:hover { transform: translateY(-5px); box-shadow: 0 12px 40px rgba(0, 0, 0, 0.15); } .card h2 { color: #2d3748; margin-bottom: 20px; font-size: 1.5em; border-bottom: 2px solid #e2e8f0; padding-bottom: 10px; } .form-group { margin-bottom: 20px; } .form-group label { display: block; margin-bottom: 8px; font-weight: 600; color: #4a5568; } .form-group input, .form-group select, .form-group textarea { width: 100%; padding: 12px 16px; border: 2px solid #e2e8f0; border-radius: 8px; font-size: 14px; transition: border-color 0.3s ease, box-shadow 0.3s ease; } .form-group input:focus, .form-group select:focus, .form-group textarea:focus { outline: none; border-color: #667eea; box-shadow: 0 0 0 3px rgba(102, 126, 234, 0.1); } .form-group textarea { resize: vertical; min-height: 100px; } .btn { background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); color: white; border: none; padding: 12px 24px; border-radius: 8px; font-size: 14px; font-weight: 600; cursor: pointer; transition: transform 0.3s ease, box-shadow 0.3s ease; margin-right: 10px; margin-bottom: 10px; } .btn:hover { transform: translateY(-2px); box-shadow: 0 4px 12px rgba(102, 126, 234, 0.3); } .btn-secondary { background: linear-gradient(135deg, #a0aec0 0%, #718096 100%); } .btn-danger { background: linear-gradient(135deg, #fc8181 0%, #f56565 100%); } .btn-success { background: linear-gradient(135deg, #48bb78 0%, #38a169 100%); } .conditions-section, .actions-section { margin-top: 20px; padding: 20px; background: #f7fafc; border-radius: 8px; border: 2px dashed #e2e8f0; } .condition-item, .action-item { display: flex; gap: 10px; margin-bottom: 10px; align-items: center; } .condition-item input, .action-item input { flex: 1; } .rules-list { max-height: 400px; overflow-y: auto; } .rule-item { background: #f7fafc; border-radius: 8px; padding: 15px; margin-bottom: 15px; border-left: 4px solid #667eea; transition: transform 0.2s ease; } .rule-item:hover { transform: translateX(5px); } .rule-item h3 { color: #2d3748; margin-bottom: 8px; font-size: 1.1em; } .rule-item p { color: #718096; margin-bottom: 10px; font-size: 0.9em; } .rule-meta { display: flex; gap: 15px; font-size: 0.8em; color: #a0aec0; } .rule-actions { margin-top: 10px; display: flex; gap: 10px; } .rule-actions .btn { padding: 6px 12px; font-size: 12px; } .filters { display: flex; gap: 15px; margin-bottom: 20px; flex-wrap: wrap; } .filters select, .filters input { padding: 8px 12px; border: 2px solid #e2e8f0; border-radius: 6px; font-size: 14px; } .stats-grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); gap: 20px; margin-bottom: 30px; } .stat-card { background: rgba(255, 255, 255, 0.95); backdrop-filter: blur(10px); border-radius: 15px; padding: 25px; text-align: center; box-shadow: 0 8px 32px rgba(0, 0, 0, 0.1); transition: transform 0.3s ease; } .stat-card:hover { transform: translateY(-5px); } .stat-number { font-size: 2.5em; font-weight: 700; color: #667eea; margin-bottom: 10px; } .stat-label { color: #718096; font-size: 1.1em; } .notification { position: fixed; top: 20px; right: 20px; padding: 15px 20px; border-radius: 8px; color: white; font-weight: 600; z-index: 1000; transform: translateX(400px); transition: transform 0.3s ease; } .notification.show { transform: translateX(0); } .notification.success { background: #48bb78; } .notification.error { background: #f56565; } .notification.info { background: #4299e1; } media (max-width: 768px) { .main-content { grid-template-columns: 1fr; } .header h1 { font-size: 2em; } .filters { flex-direction: column; } .stats-grid { grid-template-columns: 1fr; } } .loading { display: inline-block; width: 20px; height: 20px; border: 3px solid #f3f3f3; border-top: 3px solid #667eea; border-radius: 50%; animation: spin 1s linear infinite; } keyframes spin { 0% { transform: rotate(0deg); } 100% { transform: rotate(360deg); } } .empty-state { text-align: center; padding: 40px 20px; color: #a0aec0; } .empty-state i { font-size: 4em; margin-bottom: 20px; opacity: 0.5; } /style /head body div classcontainer !-- 头部 -- div classheader h1 规则定义中心/h1 p定义生成应用的规则与标准支持动态重构的备忘录仪表盘/p /div !-- 统计信息 -- div classstats-grid div classstat-card div classstat-number idtotalRules0/div div classstat-label总规则数/div /div div classstat-card div classstat-number idactiveRules0/div div classstat-label活跃规则/div /div div classstat-card div classstat-number idcategoriesCount0/div div classstat-label分类数量/div /div /div !-- 主要内容 -- div classmain-content !-- 规则创建/编辑 -- div classcard h2 创建/编辑规则/h2 form idruleForm input typehidden idruleId div classform-group label forruleName规则名称 */label input typetext idruleName required placeholder输入规则名称 /div div classform-group label forruleDescription规则描述 */label textarea idruleDescription required placeholder输入规则描述/textarea /div div classform-group label forruleCategory规则分类 */label select idruleCategory required option value选择分类/option /select /div div classform-group label forrulePriority优先级/label input typenumber idrulePriority value0 min0 max100 /div div classform-group label input typecheckbox idruleActive checked 启用规则 /label /div !-- 条件部分 -- div classconditions-section h3条件设置/h3 div idconditionsList/div button typebutton classbtn btn-secondary onclickaddCondition() 添加条件 /button /div !-- 动作部分 -- div classactions-section h3动作设置/h3 div idactionsList/div button typebutton classbtn btn-secondary onclickaddAction() 添加动作 /button /div div stylemargin-top: 20px; button typesubmit classbtn保存规则/button button typebutton classbtn btn-secondary onclickclearForm() 清空表单 /button /div /form /div !-- 规则列表 -- div classcard h2 规则列表/h2 !-- 过滤器 -- div classfilters select idcategoryFilter onchangeloadRules() option value所有分类/option /select select idstatusFilter onchangeloadRules() option value所有状态/option option valuetrue启用/option option valuefalse禁用/option /select input typetext idkeywordFilter placeholder搜索关键词 oninputloadRules() /div div classrules-list idrulesList div classempty-state div/div p暂无规则请创建新规则/p /div /div /div /div /div !-- 通知 -- div classnotification idnotification/div script let currentRuleId null; let conditionsCount 0; let actionsCount 0; // 初始化 document.addEventListener(DOMContentLoaded, function() { loadCategories(); loadRules(); loadStats(); setupForm(); }); // 设置表单 function setupForm() { document.getElementById(ruleForm).addEventListener(submit, function(e) { e.preventDefault(); saveRule(); }); } // 加载分类 function loadCategories() { fetch(/api/categories) .then(response response.json()) .then(data { const categorySelect document.getElementById(ruleCategory); const categoryFilter document.getElementById(categoryFilter); data.categories.forEach(category { const option1 document.createElement(option); option1.value category; option1.textContent category; categorySelect.appendChild(option1); const option2 document.createElement(option); option2.value category; option2.textContent category; categoryFilter.appendChild(option2); }); }) .catch(error { showNotification(加载分类失败, error); }); } // 加载规则 function loadRules() { const category document.getElementById(categoryFilter).value; const status document.getElementById(statusFilter).value; const keyword document.getElementById(keywordFilter).value; let url /api/rules?; if (category) url category${category}; if (status) url active${status}; if (keyword) url keyword${keyword}; fetch(url) .then(response response.json()) .then(data { renderRules(data.rules); }) .catch(error { showNotification(加载规则失败, error); }); } // 渲染规则列表 function renderRules(rules) { const rulesList document.getElementById(rulesList); if (rules.length 0) { rulesList.innerHTML div classempty-state div/div p没有找到符合条件的规则/p /div ; return; } rulesList.innerHTML rules.map(rule div classrule-item h3${rule.name}/h3 p${rule.description}/p div classrule-meta span分类: ${rule.category}/span span优先级: ${rule.priority}/span span状态: ${rule.active ? 启用 : 禁用}/span /div div classrule-actions button classbtn btn-secondary onclickeditRule(${rule.id}) 编辑 /button button classbtn btn-danger onclickdeleteRule(${rule.id}) 删除 /button button classbtn ${rule.active ? btn-secondary : btn-success} onclicktoggleRule(${rule.id}, ${!rule.active}) ${rule.active ? 禁用 : 启用} /button /div /div ).join(); } // 加载统计信息 function loadStats() { fetch(/api/stats) .then(response response.json()) .then(data { document.getElementById(totalRules).textContent data.stats.total_rules; document.getElementById(activeRules).textContent data.stats.active_rules; document.getElementById(categoriesCount).textContent Object.keys(data.stats.categories_count).length; }) .catch(error { console.error(加载统计信息失败:, error); }); } // 添加条件 function addCondition() { const conditionsList document.getElementById(conditionsList); const conditionId conditionsCount; const conditionDiv document.createElement(div); conditionDiv.className condition-item; conditionDiv.id condition-${conditionId}; conditionDiv.innerHTML input typetext placeholder字段名 idcondition-field-${conditionId} select idcondition-operator-${conditionId} option valueeq等于/option option valueneq不等于/option option valuegt大于/option option valuelt小于/option option valuecontains包含/option /select input typetext placeholder值 idcondition-value-${conditionId} button typebutton classbtn btn-danger onclickremoveCondition(${conditionId}) × /button ; conditionsList.appendChild(conditionDiv); } // 移除条件 function removeCondition(conditionId) { const conditionDiv document.getElementById(condition-${conditionId}); if (conditionDiv) { conditionDiv.remove(); } } // 添加动作 function addAction() { const actionsList document.getElementById(actionsList); const actionId actionsCount; const actionDiv document.createElement(div); actionDiv.className action-item; actionDiv.id action-${actionId}; actionDiv.innerHTML input typetext placeholder动作类型 idaction-type-${actionId} input typetext placeholder动作参数 idaction-param-${actionId} button typebutton classbtn btn-danger onclickremoveAction(${actionId}) × /button ; actionsList.appendChild(actionDiv); } // 移除动作 function removeAction(actionId) { const actionDiv document.getElementById(action-${actionId}); if (actionDiv) { actionDiv.remove(); } } // 获取条件数据 function getConditionsData() { const conditions []; const conditionsList document.getElementById(conditionsList).children; for (let i 0; i conditionsList.length; i) { const div conditionsList[i]; const field div.querySelector(input[id^condition-field-]).value; const operator div.querySelector(select[id^condition-operator-]).value; const value div.querySelector(input[id^condition-value-]).value; if (field value) { conditions.push({ field, operator, value }); } } return conditions; } // 获取动作数据 function getActionsData() { const actions []; const actionsList document.getElementById(actionsList).children; for (let i 0; i actionsList.length; i) { const div actionsList[i]; const type div.querySelector(input[id^action-type-]).value; const param div.querySelector(input[id^action-param-]).value; if (type) { actions.push({ type, ...(param ? { param } : {}) }); } } return actions; } // 保存规则 function saveRule() { const ruleData { name: document.getElementById(ruleName).value, description: document.getElementById(ruleDescription).value, category: document.getElementById(ruleCategory).value, priority: parseInt(document.getElementById(rulePriority).value), active: document.getElementById(ruleActive).checked, conditions: getConditionsData(), actions: getActionsData() }; const ruleId document.getElementById(ruleId).value; const url ruleId ? /api/rules/${ruleId} : /api/rules; const method ruleId ? PUT : POST; fetch(url, { method: method, headers: { Content-Type: application/json }, body: JSON.stringify(ruleData) }) .then(response response.json()) .then(data { if (data.success) { showNotification(ruleId ? 规则更新成功 : 规则创建成功, success); clearForm(); loadRules(); loadStats(); loadCategories(); } else { showNotification(data.message || 操作失败, error); } }) .catch(error { showNotification(保存规则失败, error); }); } // 编辑规则 function editRule(ruleId) { fetch(/api/rules/${ruleId}) .then(response response.json()) .then(data { if (data.rule) { const rule data.rule; document.getElementById(ruleId).value rule.id; document.getElementById(ruleName).value rule.name; document.getElementById(ruleDescription).value rule.description; document.getElementById(ruleCategory).value rule.category; document.getElementById(rulePriority).value rule.priority || 0; document.getElementById(ruleActive).checked rule.active; // 清空现有条件和动作 document.getElementById(conditionsList).innerHTML ; document.getElementById(actionsList).innerHTML ; conditionsCount 0; actionsCount 0; // 添加条件 rule.conditions.forEach(condition { addCondition(); const conditionId conditionsCount - 1; document.getElementById(condition-field-${conditionId}).value condition.field; document.getElementById(condition-operator-${conditionId}).value condition.operator; document.getElementById(condition-value-${conditionId}).value condition.value; }); // 添加动作 rule.actions.forEach(action { addAction(); const actionId actionsCount - 1; document.getElementById(action-type-${actionId}).value action.type; if (action.param) { document.getElementById(action-param-${actionId}).value action.param; } }); showNotification(规则加载成功, info); } }) .catch(error { showNotification(加载规则失败, error); }); } // 删除规则 function deleteRule(ruleId) { if (confirm(确定要删除这个规则吗)) { fetch(/api/rules/${ruleId}, { method: DELETE }) .then(response response.json()) .then(data { if (data.success) { showNotification(规则删除成功, success); loadRules(); loadStats(); loadCategories(); } else { showNotification(data.message || 删除失败, error); } }) .catch(error { showNotification(删除规则失败, error); }); } } // 切换规则状态 function toggleRule(ruleId, active) { fetch(/api/rules/${ruleId}, { method: PUT, headers: { Content-Type: application/json }, body: JSON.stringify({ active: active }) }) .then(response response.json()) .then(data { if (data.success) { showNotification(规则已${active ? 启用 : 禁用}, success); loadRules(); loadStats(); } else { showNotification(data.message || 操作失败, error); } }) .catch(error { showNotification(操作失败, error); }); } // 清空表单 function clearForm() { document.getElementById(ruleForm).reset(); document.getElementById(ruleId).value ; document.getElementById(conditionsList).innerHTML ; document.getElementById(actionsList).innerHTML ; conditionsCount 0; actionsCount 0; } // 显示通知 function showNotification(message, type info) { const notification document.getElementById(notification); notification.textContent message; notification.className notification ${type} show; setTimeout(() { notification.classList.remove(show); }, 3000); } // 自动刷新 setInterval(() { loadStats(); }, 30000); // 每30秒刷新一次统计信息 /script /body /htmlapp_run_app.py#!/usr/bin/env python3 # -*- coding: utf-8 -*- 应用运行HTTP服务 提供RESTful API接口来管理应用的启动、停止和状态监控 import sys import os import json import requests from http.server import HTTPServer, BaseHTTPRequestHandler from urllib.parse import parse_qs, urlparse # 添加lib目录到路径 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from lib.run_lib import AppRunner class RunHTTPHandler(BaseHTTPRequestHandler): 应用运行HTTP处理器 def __init__(self, *args, **kwargs): 初始化处理器 # 获取正确的数据库路径 base_dir os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) db_path os.path.join(base_dir, data, running_apps.json) self.runner AppRunner(db_path) # 应用生成服务的地址 self.generate_service_url http://localhost:8002 super().__init__(*args, **kwargs) def do_GET(self): 处理GET请求 parsed_url urlparse(self.path) path parsed_url.path query_params parse_qs(parsed_url.query) try: if path /api/apps: # 获取运行中的应用列表 running_apps self.runner.get_running_apps() # 如果请求包含所有应用包括未运行的 if all in query_params and query_params[all][0].lower() true: all_apps self._get_all_apps() self._send_response(200, {apps: all_apps, running: running_apps, total: len(all_apps)}) else: self._send_response(200, {apps: running_apps, total: len(running_apps)}) elif path.startswith(/api/apps/): app_id path.split(/)[-1] if /health in path: # 获取应用健康状态 health self.runner.check_app_health(app_id) self._send_response(200, {health: health}) elif /logs in path: # 获取应用日志 log_type query_params.get(type, [stdout])[0] limit int(query_params.get(limit, [100])[0]) logs self.runner.get_app_logs(app_id, log_type, limit) self._send_response(200, {logs: logs, type: log_type, count: len(logs)}) else: # 获取单个应用状态 app_status self.runner.get_app_status(app_id) if app_status: self._send_response(200, {app: app_status}) else: # 检查应用是否存在但未运行 app_info self._get_app_info(app_id) if app_info: self._send_response(200, {app: app_info, status: stopped}) else: self._send_response(404, {error: 应用不存在}) elif path /api/stats: # 获取统计信息 stats self.runner.get_runner_stats() self._send_response(200, {stats: stats}) elif path /: # 提供前端页面 self._serve_static_file(page/app_run.html) else: # 尝试提供静态文件 self._serve_static_file(path.lstrip(/)) except Exception as e: self._send_response(500, {error: f服务器内部错误: {str(e)}}) def do_POST(self): 处理POST请求 parsed_url urlparse(self.path) path parsed_url.path try: content_length int(self.headers.get(Content-Length, 0)) body self.rfile.read(content_length).decode(utf-8) data json.loads(body) if body else {} if path.startswith(/api/apps/) and /start in path: # 启动应用 app_id path.split(/)[-2] # 获取应用信息 app_info self._get_app_info(app_id) if not app_info: self._send_response(404, {error: 应用不存在}) return # 启动应用 result self.runner.start_app(app_id, app_info) if result[success]: self._send_response(200, result) else: self._send_response(400, result) elif path.startswith(/api/apps/) and /stop in path: # 停止应用 app_id path.split(/)[-2] result self.runner.stop_app(app_id) if result[success]: self._send_response(200, result) else: self._send_response(400, result) elif path.startswith(/api/apps/) and /restart in path: # 重启应用 app_id path.split(/)[-2] # 获取应用信息 app_info self._get_app_info(app_id) if not app_info: self._send_response(404, {error: 应用不存在}) return # 重启应用 result self.runner.restart_app(app_id, app_info) if result[success]: self._send_response(200, result) else: self._send_response(400, result) else: self._send_response(404, {error: 接口不存在}) except json.JSONDecodeError: self._send_response(400, {error: 无效的JSON格式}) except Exception as e: self._send_response(500, {error: f服务器内部错误: {str(e)}}) def _get_all_apps(self) - list: 获取所有应用包括未运行的 try: # 从应用生成服务获取所有应用 response requests.get(f{self.generate_service_url}/api/apps) if response.status_code 200: all_apps response.json().get(apps, []) # 获取运行中的应用 running_apps self.runner.get_running_apps() running_ids {app[id] for app in running_apps} # 合并应用信息 for app in all_apps: if app[id] in running_ids: app[status] running else: app[status] stopped return all_apps return [] except Exception: return [] def _get_app_info(self, app_id: str) - dict: 获取应用信息 try: response requests.get(f{self.generate_service_url}/api/apps/{app_id}) if response.status_code 200: return response.json().get(app, {}) return {} except Exception: return {} def _send_response(self, status_code: int, data: dict): 发送JSON响应 self.send_response(status_code) self.send_header(Content-Type, application/json) self.send_header(Access-Control-Allow-Origin, *) self.send_header(Access-Control-Allow-Methods, GET, POST, PUT, DELETE, OPTIONS) self.send_header(Access-Control-Allow-Headers, Content-Type, Authorization) self.end_headers() response_body json.dumps(data, ensure_asciiFalse).encode(utf-8) self.wfile.write(response_body) def _serve_static_file(self, file_path: str): 提供静态文件 base_dir os.path.dirname(os.path.dirname(os.path.abspath(__file__))) full_path os.path.join(base_dir, file_path) # 安全检查防止目录遍历 if not full_path.startswith(base_dir): self._send_response(403, {error: 访问被拒绝}) return # 检查文件是否存在 if not os.path.exists(full_path) or not os.path.isfile(full_path): self._send_response(404, {error: 文件不存在}) return # 确定MIME类型 mime_type self._get_mime_type(full_path) # 发送文件 self.send_response(200) self.send_header(Content-Type, mime_type) self.send_header(Access-Control-Allow-Origin, *) self.end_headers() with open(full_path, rb) as f: self.wfile.write(f.read()) def _get_mime_type(self, file_path: str) - str: 获取文件的MIME类型 extensions { .html: text/html, .css: text/css, .js: application/javascript, .json: application/json, .png: image/png, .jpg: image/jpeg, .jpeg: image/jpeg, .gif: image/gif, .svg: image/svgxml, .txt: text/plain } _, ext os.path.splitext(file_path.lower()) return extensions.get(ext, application/octet-stream) def do_OPTIONS(self): 处理OPTIONS请求 self.send_response(200) self.send_header(Access-Control-Allow-Origin, *) self.send_header(Access-Control-Allow-Methods, GET, POST, PUT, DELETE, OPTIONS) self.send_header(Access-Control-Allow-Headers, Content-Type, Authorization) self.end_headers() def run_server(port: int 8003): 启动HTTP服务器 server_address (, port) httpd HTTPServer(server_address, RunHTTPHandler) print(f应用运行服务启动成功监听端口: {port}) print(f访问地址: http://localhost:{port}) print(fAPI文档: http://localhost:{port}/api) try: httpd.serve_forever() except KeyboardInterrupt: print(\n服务器正在关闭...) httpd.server_close() print(服务器已关闭) if __name__ __main__: # 启动服务器 run_serverrun_lib.py#!/usr/bin/env python3 # -*- coding: utf-8 -*- 应用运行核心库 - 管理应用的启动、停止和状态监控 支持动态应用管理和实时状态监控 import json import os import sys import uuid import subprocess import time import signal import psutil from datetime import datetime from typing import Dict, List, Any, Optional class AppRunner: 应用运行管理器 - 管理应用的启动、停止和状态监控 def __init__(self, db_path: str ../data/running_apps.json): 初始化应用运行管理器 self.db_path os.path.abspath(os.path.join(os.path.dirname(__file__), db_path)) self.running_apps self._load_running_apps() self.base_dir os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) def _load_running_apps(self) - Dict[str, Any]: 加载运行中的应用数据 try: if os.path.exists(self.db_path): with open(self.db_path, r, encodingutf-8) as f: return json.load(f) return {apps: {}, version: 1.0 except Exception as e: print(f加载运行应用数据失败: {e}) return {apps: {}, version: 1.0 def _save_running_apps(self): 保存运行中的应用数据 try: os.makedirs(os.path.dirname(self.db_path), exist_okTrue) with open(self.db_path, w, encodingutf-8) as f: json.dump(self.running_apps, f, ensure_asciiFalse, indent2) except Exception as e: print(f保存运行应用数据失败: {e}) raise def start_app(self, app_id: str, app_info: Dict[str, Any]) - Dict[str, Any]: 启动应用 app_info 应该包含应用的基本信息如目录、端口等 try: # 检查应用是否已经在运行 if app_id in self.running_apps[apps]: return {success: False, message: 应用已经在运行} # 检查应用目录是否存在 app_dir app_info.get(directory) if not app_dir or not os.path.exists(app_dir): return {success: False, message: 应用目录不存在} # 检查启动脚本是否存在 start_script os.path.join(app_dir, start.sh) if not os.path.exists(start_script): return {success: False, message: 启动脚本不存在} # 启动应用 print(f正在启动应用 {app_id}...) # 使用 subprocess.Popen 启动应用 process subprocess.Popen( [bash, start_script], cwdapp_dir, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue ) # 等待应用启动给应用一些启动时间 time.sleep(2) # 检查应用是否成功启动 if process.poll() is not None: # 进程已经结束说明启动失败 stderr process.stderr.read() if process.stderr else return {success: False, message: f应用启动失败: {stderr}} # 保存应用运行信息 running_info { id: app_id, name: app_info.get(name, Unknown), description: app_info.get(description, ), type: app_info.get(type, unknown), port: app_info.get(port, 0), directory: app_dir, pid: process.pid, start_time: datetime.now().isoformat(), status: running, process: process.pid, # 保存进程ID stdout: [], stderr: [] } self.running_apps[apps][app_id] running_info self._save_running_apps() print(f应用 {app_id} 启动成功PID: {process.pid}) return { success: True, app_id: app_id, pid: process.pid, port: app_info.get(port, 0), message: 应用启动成功 } except Exception as e: return {success: False, message: f应用启动失败: {str(e)}} def stop_app(self, app_id: str) - Dict[str, Any]: 停止应用 try: if app_id not in self.running_apps[apps]: return {success: False, message: 应用未在运行} app_info self.running_apps[apps][app_id] pid app_info.get(pid) if not pid: return {success: False, message: 应用进程ID不存在} print(f正在停止应用 {app_id} (PID: {pid})...) try: # 尝试优雅地终止进程 os.kill(pid, signal.SIGTERM) # 等待进程结束 timeout 5 start_time time.time() while time.time() - start_time timeout: if not psutil.pid_exists(pid): break time.sleep(0.5) # 如果进程仍然存在强制终止 if psutil.pid_exists(pid): os.kill(pid, signal.SIGKILL) print(f应用 {app_id} 已被强制终止) else: print(f应用 {app_id} 已正常终止) except psutil.NoSuchProcess: print(f进程 {pid} 不存在可能已经终止) except Exception as e: print(f终止进程时出错: {e}) # 从运行列表中移除 del self.running_apps[apps][app_id] self._save_running_apps() return {success: True, message: 应用停止成功} except Exception as e: return {success: False, message: f应用停止失败: {str(e)}} def restart_app(self, app_id: str, app_info: Dict[str, Any]) - Dict[str, Any]: 重启应用 try: # 先停止应用如果在运行 if app_id in self.running_apps[apps]: stop_result self.stop_app(app_id) if not stop_result[success]: return {success: False, message: f停止应用失败: {stop_result[message]}} # 等待一会儿再启动 time.sleep(1) # 启动应用 return self.start_app(app_id, app_info) except Exception as e: return {success: False, message: f应用重启失败: {str(e)}} def get_running_apps(self) - List[Dict[str, Any]]: 获取所有运行中的应用 try: # 更新应用状态 self._update_apps_status() return list(self.running_apps[apps].values()) except Exception as e: print(f获取运行应用列表失败: {e}) return [] def get_app_status(self, app_id: str) - Optional[Dict[str, Any]]: 获取单个应用的状态 try: if app_id not in self.running_apps[apps]: return None # 更新应用状态 self._update_app_status(app_id) return self.running_apps[apps][app_id] except Exception as e: print(f获取应用状态失败: {e}) return None def _update_apps_status(self): 更新所有应用的状态 apps_to_remove [] for app_id, app_info in self.running_apps[apps].items(): pid app_info.get(pid) if pid and not psutil.pid_exists(pid): # 进程不存在应用已经停止 app_info[status] stopped app_info[end_time] datetime.now().isoformat() apps_to_remove.append(app_id) # 移除已经停止的应用 for app_id in apps_to_remove: del self.running_apps[apps][app_id] if apps_to_remove: self._save_running_apps() def _update_app_status(self, app_id: str): 更新单个应用的状态 if app_id not in self.running_apps[apps]: return app_info self.running_apps[apps][app_id] pid app_info.get(pid) if pid and not psutil.pid_exists(pid): # 进程不存在应用已经停止 app_info[status] stopped app_info[end_time] datetime.now().isoformat() del self.running_apps[apps][app_id] self._save_running_apps() def get_app_logs(self, app_id: str, log_type: str stdout, limit: int 100) - List[str]: 获取应用日志 try: if app_id not in self.running_apps[apps]: return [] app_info self.running_apps[apps][app_id] if log_type stdout: return app_info.get(stdout, [])[-limit:] elif log_type stderr: return app_info.get(stderr, [])[-limit:] return [] except Exception as e: print(f获取应用日志失败: {e}) return [] def check_app_health(self, app_id: str) - Dict[str, Any]: 检查应用健康状态 try: if app_id not in self.running_apps[apps]: return { healthy: False, status: not_running, message: 应用未在运行 } app_info self.running_apps[apps][app_id] pid app_info.get(pid) if not pid or not psutil.pid_exists(pid): return { healthy: False, status: crashed, message: 应用进程不存在 } # 检查端口是否在监听 port app_info.get(port) if port: try: # 尝试连接到应用端口 import socket sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(2) result sock.connect_ex((localhost, port)) sock.close() if result ! 0: return { healthy: False, status: port_closed, message: f应用端口 {port} 未在监听 } except Exception: pass return { healthy: True, status: running, pid: pid, port: port, uptime: self._calculate_uptime(app_info.get(start_time)), message: 应用运行正常 } except Exception as e: return { healthy: False, status: error, message: f健康检查失败: {str(e)} } def _calculate_uptime(self, start_time_str: str) - str: 计算运行时间 try: start_time datetime.fromisoformat(start_time_str) uptime datetime.now() - start_time days uptime.days hours, remainder divmod(uptime.seconds, 3600) minutes, seconds divmod(remainder, 60) parts [] if days 0: parts.append(f{days}天) if hours 0: parts.append(f{hours}小时) if minutes 0: parts.append(f{minutes}分钟) if seconds 0 or not parts: parts.append(f{seconds}秒) return .join(parts) except Exception: return 未知 def get_runner_stats(self) - Dict[str, Any]: 获取运行管理器统计信息 try: # 更新应用状态 self._update_apps_status() running_apps list(self.running_apps[apps].values()) total_running len(running_apps) # 按类型统计 type_count {} for app in running_apps: app_type app.get(type, unknown) type_count[app_type] type_count.get(app_type, 0) 1 return { total_running: total_running, type_count: type_count, last_updated: datetime.now().isoformat(), apps: running_apps } except Exception as e: print(f获取运行统计失败: {e}) return { total_running: 0, type_count: {}, last_updated: datetime.now().isoformat(), apps: [] } def cleanup_stopped_apps(self): 清理已停止的应用 self._update_apps_status() print(已清理停止的应用) # 示例使用 if __name__ __main__: # 创建应用运行管理器实例 runner AppRunner() print(应用运行管理器示例) print(1. 获取运行统计) print(2. 清理停止的应用) choice input(请选择操作: ) if choice 1: stats runner.get_runner_stats() print(f运行统计: {stats}) elif choice 2: runner.cleanup_stopped_apps() print(清理完成) else: print(无效选择app_run.html!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title应用运行中心 - 多子项目协同中枢/title style * { margin: 0; padding: 0; box-sizing: border-box; } body { font-family: -apple-system, BlinkMacSystemFont, Segoe UI, Roboto, Helvetica Neue, Arial, sans-serif; background: linear-gradient(135deg, #4facfe 0%, #00f2fe 100%); min-height: 100vh; color: #333; } .container { max-width: 1400px; margin: 0 auto; padding: 20px; } .header { background: rgba(255, 255, 255, 0.95); backdrop-filter: blur(10px); border-radius: 15px; padding: 30px; margin-bottom: 30px; box-shadow: 0 8px 32px rgba(0, 0, 0, 0.1); text-align: center; } .header h1 { font-size: 2.5em; color: #4a5568; margin-bottom: 10px; font-weight: 700; } .header p { color: #718096; font-size: 1.1em; } .main-content { display: grid; grid-template-columns: 1fr 350px; gap: 30px; margin-bottom: 30px; } .card { background: rgba(255, 255, 255, 0.95); backdrop-filter: blur(10px); border-radius: 15px; padding: 30px; box-shadow: 0 8px 32px rgba(0, 0, 0, 0.1); transition: transform 0.3s ease, box-shadow 0.3s ease; } .card:hover { transform: translateY(-5px); box-shadow: 0 12px 40px rgba(0, 0, 0, 0.15); } .card h2 { color: #2d3748; margin-bottom: 20px; font-size: 1.5em; border-bottom: 2px solid #e2e8f0; padding-bottom: 10px; display: flex; align-items: center; gap: 10px; } .app-grid { display: grid; grid-template-columns: repeat(auto-fill, minmax(300px, 1fr)); gap: 20px; margin-bottom: 20px; } .app-card { background: #f7fafc; border-radius: 12px; padding: 20px; border-left: 4px solid #4facfe; transition: all 0.3s ease; position: relative; overflow: hidden; } .app-card:hover { transform: translateY(-3px); box-shadow: 0 8px 25px rgba(0, 0, 0, 0.1); } .app-card.running { border-left-color: #48bb78; background: linear-gradient(135deg, #f0fff4 0%, #f7fafc 100%); } .app-card.stopped { border-left-color: #a0aec0; background: #f7fafc; } .app-header { display: flex; justify-content: space-between; align-items: flex-start; margin-bottom: 15px; } .app-title { flex: 1; } .app-title h3 { color: #2d3748; font-size: 1.2em; margin-bottom: 5px; display: flex; align-items: center; gap: 8px; } .app-title p { color: #718096; font-size: 0.9em; line-height: 1.4; } .app-status { position: absolute; top: 15px; right: 15px; padding: 4px 12px; border-radius: 20px; font-size: 0.8em; font-weight: 600; text-transform: uppercase; } .app-status.running { background: #c6f6d5; color: #22543d; } .app-status.stopped { background: #e2e8f0; color: #4a5568; } .app-meta { display: grid; grid-template-columns: 1fr 1fr; gap: 10px; margin-bottom: 15px; font-size: 0.85em; } .meta-item { display: flex; align-items: center; gap: 5px; color: #718096; } .meta-item .label { font-weight: 600; color: #4a5568; } .app-actions { display: flex; gap: 8px; flex-wrap: wrap; } .btn { background: linear-gradient(135deg, #4facfe 0%, #00f2fe 100%); color: white; border: none; padding: 8px 16px; border-radius: 6px; font-size: 12px; font-weight: 600; cursor: pointer; transition: all 0.3s ease; display: flex; align-items: center; gap: 5px; } .btn:hover { transform: translateY(-1px); box-shadow: 0 4px 12px rgba(79, 172, 254, 0.3); } .btn-success { background: linear-gradient(135deg, #48bb78 0%, #38a169 100%); } .btn-success:hover { box-shadow: 0 4px 12px rgba(72, 187, 120, 0.3); } .btn-danger { background: linear-gradient(135deg, #fc8181 0%, #f56565 100%); } .btn-danger:hover { box-shadow: 0 4px 12px rgba(252, 129, 129, 0.3); } .btn-secondary { background: linear-gradient(135deg, #a0aec0 0%, #718096 100%); } .btn-secondary:hover { box-shadow: 0 4px 12px rgba(160, 174, 192, 0.3); } .btn-info { background: linear-gradient(135deg, #4299e1 0%, #3182ce 100%); } .btn-info:hover { box-shadow: 0 4px 12px rgba(66, 153, 225, 0.3); } .btn:disabled { opacity: 0.6; cursor: not-allowed; transform: none; } .btn:disabled:hover { box-shadow: none; } .filters { display: flex; gap: 15px; margin-bottom: 20px; flex-wrap: wrap; padding: 20px; background: #f7fafc; border-radius: 8px; } .filters select, .filters input { padding: 8px 12px; border: 2px solid #e2e8f0; border-radius: 6px; font-size: 14px; transition: border-color 0.3s ease; } .filters select:focus, .filters input:focus { outline: none; border-color: #4facfe; } .stats-grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(120px, 1fr)); gap: 15px; margin-bottom: 20px; } .stat-box { background: linear-gradient(135deg, #4facfe 0%, #00f2fe 100%); color: white; padding: 20px; border-radius: 10px; text-align: center; transition: transform 0.3s ease; } .stat-box:hover { transform: scale(1.05); } .stat-box.running { background: linear-gradient(135deg, #48bb78 0%, #38a169 100%); } .stat-box.stopped { background: linear-gradient(135deg, #a0aec0 0%, #718096 100%); } .stat-number { font-size: 2em; font-weight: 700; margin-bottom: 5px; } .stat-label { font-size: 0.9em; opacity: 0.9; } .detail-panel { background: #f7fafc; border-radius: 8px; padding: 20px; margin-bottom: 20px; border-left: 4px solid #4facfe; } .detail-panel h4 { color: #2d3748; margin-bottom: 15px; font-size: 1.1em; } .detail-item { display: flex; justify-content: space-between; margin-bottom: 10px; padding: 8px 0; border-bottom: 1px solid #e2e8f0; } .detail-item:last-child { border-bottom: none; } .detail-label { font-weight: 600; color: #4a5568; } .detail-value { color: #718096; } .logs-container { max-height: 300px; overflow-y: auto; background: #1a202c; border-radius: 6px; padding: 15px; font-family: Courier New, monospace; font-size: 0.9em; line-height: 1.4; } .log-entry { color: #e2e8f0; margin-bottom: 5px; padding-left: 20px; position: relative; } .log-entry:before { content: ; position: absolute; left: 0; color: #4facfe; } .log-entry.error { color: #fc8181; } .log-entry.error:before { color: #fc8181; content: !; } .notification { position: fixed; top: 20px; right: 20px; padding: 15px 20px; border-radius: 8px; color: white; font-weight: 600; z-index: 1000; transform: translateX(400px); transition: transform 0.3s ease; box-shadow: 0 4px 12px rgba(0, 0, 0, 0.2); } .notification.show { transform: translateX(0); } .notification.success { background: linear-gradient(135deg, #48bb78 0%, #38a169 100%); } .notification.error { background: linear-gradient(135deg, #fc8181 0%, #f56565 100%); } .notification.info { background: linear-gradient(135deg, #4299e1 0%, #3182ce 100%); } .loading { display: inline-block; width: 16px; height: 16px; border: 2px solid #f3f3f3; border-top: 2px solid #4facfe; border-radius: 50%; animation: spin 1s linear infinite; } keyframes spin { 0% { transform: rotate(0deg); } 100% { transform: rotate(360deg); } } .empty-state { text-align: center; padding: 60px 20px; color: #a0aec0; } .empty-state i { font-size: 4em; margin-bottom: 20px; opacity: 0.5; } .refresh-btn { position: fixed; bottom: 30px; right: 30px; width: 50px; height: 50px; border-radius: 50%; background: linear-gradient(135deg, #4facfe 0%, #00f2fe 100%); color: white; border: none; font-size: 1.2em; cursor: pointer; box-shadow: 0 4px 20px rgba(79, 172, 254, 0.4); transition: all 0.3s ease; z-index: 100; } .refresh-btn:hover { transform: scale(1.1); box-shadow: 0 6px 25px rgba(79, 172, 254, 0.6); } .refresh-btn:active { transform: scale(0.95); } .refresh-btn.loading { background: linear-gradient(135deg, #a0aec0 0%, #718096 100%); } media (max-width: 768px) { .main-content { grid-template-columns: 1fr; } .app-grid { grid-template-columns: 1fr; } .stats-grid { grid-template-columns: repeat(2, 1fr); } .header h1 { font-size: 2em; } .filters { flex-direction: column; } } /style /head body div classcontainer !-- 头部 -- div classheader h1⚙️ 应用运行中心/h1 p管理应用的启动、停止和状态监控实时掌握应用运行情况/p /div !-- 主要内容 -- div classmain-content !-- 左侧应用管理 -- div !-- 统计信息 -- div classcard h2 运行统计/h2 div classstats-grid div classstat-box div classstat-number idtotalApps0/div div classstat-label总应用/div /div div classstat-box running div classstat-number idrunningApps0/div div classstat-label运行中/div /div div classstat-box stopped div classstat-number idstoppedApps0/div div classstat-label已停止/div /div div classstat-box div classstat-number idtotalUptime0/div div classstat-label总运行时/div /div /div /div !-- 过滤器 -- div classcard stylemargin-top: 20px; h2 应用筛选/h2 div classfilters select idstatusFilter onchangeloadApps() option value所有状态/option option valuerunning运行中/option option valuestopped已停止/option /select select idtypeFilter onchangeloadApps() option value所有类型/option /select input typetext idkeywordFilter placeholder搜索应用名称或描述 oninputloadApps() button classbtn btn-secondary onclickloadApps() span刷新/span span idfilterLoading classloading styledisplay: none;/span /button /div /div !-- 应用列表 -- div classcard stylemargin-top: 20px; h2 应用列表/h2 div classapp-grid idappGrid div classempty-state div/div p暂无应用数据/p /div /div /div /div !-- 右侧详情面板 -- div div classcard h2 应用详情/h2 div idappDetail classdetail-panel styledisplay: none; h4 iddetailTitle应用详情/h4 div iddetailContent !-- 详情内容将通过JavaScript动态填充 -- /div /div div classempty-state iddetailEmpty stylemargin-top: 20px; div/div p请选择一个应用查看详情/p /div /div !-- 健康状态 -- div classcard stylemargin-top: 20px; h2 健康状态/h2 div idhealthStatus classdetail-panel styledisplay: none; h4健康检查/h4 div idhealthContent !-- 健康状态内容将通过JavaScript动态填充 -- /div /div div classempty-state idhealthEmpty stylemargin-top: 20px; div/div p暂无健康状态数据/p /div /div !-- 应用日志 -- div classcard stylemargin-top: 20px; h2 应用日志/h2 div idappLogs styledisplay: none; div stylemargin-bottom: 15px; button classbtn btn-info onclickloadLogs(stdout) idstdoutBtn 标准输出 /button button classbtn btn-secondary onclickloadLogs(stderr) idstderrBtn 错误输出 /button /div div classlogs-container idlogsContainer !-- 日志内容将通过JavaScript动态填充 -- /div /div div classempty-state idlogsEmpty stylemargin-top: 20px; div/div p暂无日志数据/p /div /div /div /div /div !-- 通知 -- div classnotification idnotification/div !-- 刷新按钮 -- button classrefresh-btn onclickrefreshAll() title刷新所有数据 /button script let selectedAppId null; let currentLogType stdout; let refreshInterval null; // 应用类型图标映射 const appTypeIcons { simple_http: , data_processor: }; // 初始化 document.addEventListener(DOMContentLoaded, function() { loadApps(); loadStats(); setupAutoRefresh(); }); // 设置自动刷新 function setupAutoRefresh() { // 每30秒自动刷新一次数据 refreshInterval setInterval(() { if (selectedAppId) { loadAppDetail(selectedAppId); loadHealthStatus(selectedAppId); loadLogs(currentLogType); } loadStats(); }, 30000); } // 加载应用列表 function loadApps() { const status document.getElementById(statusFilter).value; const type document.getElementById(typeFilter).value; const keyword document.getElementById(keywordFilter).value; // 显示加载状态 document.getElementById(filterLoading).style.display inline-block; fetch(/api/apps?alltrue) .then(response response.json()) .then(data { // 隐藏加载状态 document.getElementById(filterLoading).style.display none; // 应用过滤 let filteredApps data.apps; if (status) { filteredApps filteredApps.filter(app app.status status); } if (type) { filteredApps filteredApps.filter(app app.type type); } if (keyword) { const lowerKeyword keyword.toLowerCase(); filteredApps filteredApps.filter(app app.name.toLowerCase().includes(lowerKeyword) || app.description.toLowerCase().includes(lowerKeyword) ); } renderApps(filteredApps); updateTypeFilter(data.apps); }) .catch(error { document.getElementById(filterLoading).style.display none; showNotification(加载应用列表失败, error); }); } // 更新类型过滤器 function updateTypeFilter(apps) { const typeFilter document.getElementById(typeFilter); const types [...new Set(apps.map(app app.type))]; // 清空现有选项保留第一个所有类型选项 while (typeFilter.options.length 1) { typeFilter.remove(1); } // 添加新的类型选项 types.forEach(type { const option document.createElement(option); option.value type; option.textContent type; typeFilter.appendChild(option); }); } // 渲染应用列表 function renderApps(apps) { const appGrid document.getElementById(appGrid); if (apps.length 0) { appGrid.innerHTML div classempty-state div/div p没有找到符合条件的应用/p /div ; return; } appGrid.innerHTML apps.map(app div classapp-card ${app.status} onclickselectApp(${app.id}) div classapp-header div classapp-title h3 ${appTypeIcons[app.type] || } ${app.name} /h3 p${app.description}/p /div span classapp-status ${app.status} ${app.status running ? 运行中 : 已停止} /span /div div classapp-meta div classmeta-item span classlabel类型:/span span${app.type}/span /div div classmeta-item span classlabel端口:/span span${app.port}/span /div div classmeta-item span classlabelPID:/span span${app.pid || -}/span /div div classmeta-item span classlabel运行时:/span span${app.status running ? calculateUptime(app.start_time) : -}/span /div /div div classapp-actions ${app.status running ? button classbtn btn-danger onclickevent.stopPropagation(); stopApp(${app.id}) 停止 /button button classbtn btn-secondary onclickevent.stopPropagation(); restartApp(${app.id}) 重启 /button button classbtn btn-info onclickevent.stopPropagation(); viewApp(${app.port}) ️ 查看 /button : button classbtn btn-success onclickevent.stopPropagation(); startApp(${app.id}) 启动 /button } /div /div ).join(); } // 选择应用 function selectApp(appId) { selectedAppId appId; loadAppDetail(appId); loadHealthStatus(appId); loadLogs(stdout); // 高亮选中的应用卡片 document.querySelectorAll(.app-card).forEach(card { card.style.borderLeftWidth card.onclick.toString().includes(appId) ? 6px : 4px; }); } // 加载应用详情 function loadAppDetail(appId) { fetch(/api/apps/${appId}) .then(response response.json()) .then(data { const app data.app; if (!app) { return; } // 显示详情面板隐藏空状态 document.getElementById(appDetail).style.display block; document.getElementById(detailEmpty).style.display none; // 更新详情标题 document.getElementById(detailTitle).textContent ${app.name} - 详情; // 填充详情内容 document.getElementById(detailContent).innerHTML div classdetail-item span classdetail-label应用ID:/span span classdetail-value${app.id}/span /div div classdetail-item span classdetail-label应用名称:/span span classdetail-value${app.name}/span /div div classdetail-item span classdetail-label应用类型:/span span classdetail-value${app.type}/span /div div classdetail-item span classdetail-label端口:/span span classdetail-value${app.port}/span /div div classdetail-item span classdetail-label状态:/span span classdetail-value span classapp-status ${app.status} ${app.status running ? 运行中 : 已停止} /span /span /div div classdetail-item span classdetail-label进程ID:/span span classdetail-value${app.pid || -}/span /div div classdetail-item span classdetail-label启动时间:/span span classdetail-value${app.start_time ? formatTime(app.start_time) : -}/span /div div classdetail-item span classdetail-label运行时长:/span span classdetail-value${app.status running app.start_time ? calculateUptime(app.start_time) : -}/span /div ; }) .catch(error { console.error(加载应用详情失败:, error); }); } // 加载健康状态 function loadHealthStatus(appId) { fetch(/api/apps/${appId}/health) .then(response response.json()) .then(data { const health data.health; // 显示健康状态面板隐藏空状态 document.getElementById(healthStatus).style.display block; document.getElementById(healthEmpty).style.display none; // 填充健康状态内容 document.getElementById(healthContent).innerHTML div classdetail-item span classdetail-label健康状态:/span span classdetail-value span classapp-status ${health.healthy ? running : stopped} ${health.healthy ? 健康 : 异常} /span /span /div div classdetail-item span classdetail-label状态:/span span classdetail-value${health.status}/span /div div classdetail-item span classdetail-label消息:/span span classdetail-value${health.message}/span /div ${health.pid ? div classdetail-item span classdetail-label进程ID:/span span classdetail-value${health.pid}/span /div : } ${health.uptime ? div classdetail-item span classdetail-label运行时长:/span span classdetail-value${health.uptime}/span /div : } ; }) .catch(error { console.error(加载健康状态失败:, error); }); } // 加载日志 function loadLogs(logType) { if (!selectedAppId) { return; } currentLogType logType; // 更新按钮状态 document.getElementById(stdoutBtn).className logType stdout ? btn btn-info : btn btn-secondary; document.getElementById(stderrBtn).className logType stderr ? btn btn-info : btn btn-secondary; fetch(/api/apps/${selectedAppId}/logs?type${logType}limit50) .then(response response.json()) .then(data { const logs data.logs; // 显示日志面板隐藏空状态 document.getElementById(appLogs).style.display block; document.getElementById(logsEmpty).style.display none; // 填充日志内容 const logsContainer document.getElementById(logsContainer); if (logs.length 0) { logsContainer.innerHTML div classlog-entry暂无${logType stdout ? 标准输出 : 错误输出}日志/div ; } else { logsContainer.innerHTML logs.map(log div classlog-entry ${logType stderr ? error : } ${log} /div ).join(); // 滚动到底部 logsContainer.scrollTop logsContainer.scrollHeight; } }) .catch(error { console.error(加载日志失败:, error); }); } // 启动应用 function startApp(appId) { fetch(/api/apps/${appId}/start, { method: POST, headers: { Content-Type: application/json }, body: {} }) .then(response response.json()) .then(data { if (data.success) { showNotification(应用启动成功, success); loadApps(); if (selectedAppId appId) { loadAppDetail(appId); loadHealthStatus(appId); } } else { showNotification(data.message || 启动失败, error); } }) .catch(error { showNotification(启动应用失败, error); }); } // 停止应用 function stopApp(appId) { if (confirm(确定要停止这个应用吗)) { fetch(/api/apps/${appId}/stop, { method: POST, headers: { Content-Type: application/json }, body: {} }) .then(response response.json()) .then(data { if (data.success) { showNotification(应用停止成功, success); loadApps(); if (selectedAppId appId) { loadAppDetail(appId); loadHealthStatus(appId); } } else { showNotification(data.message || 停止失败, error); } }) .catch(error { showNotification(停止应用失败, error); }); } } // 重启应用 function restartApp(appId) { if (confirm(确定要重启这个应用吗)) { fetch(/api/apps/${appId}/restart, { method: POST, headers: { Content-Type: application/json }, body: {} }) .then(response response.json()) .then(data { if (data.success) { showNotification(应用重启成功, success); loadApps(); if (selectedAppId appId) { loadAppDetail(appId); loadHealthStatus(appId); } } else { showNotification(data.message || 重启失败, error); } }) .catch(error { showNotification(重启应用失败, error); }); } } // 查看应用 function viewApp(port) { window.open(http://localhost:${port}, _blank); } // 加载统计信息 function loadStats() { fetch(/api/stats) .then(response response.json()) .then(data { const stats data.stats; document.getElementById(totalApps).textContent stats.total_running || 0; document.getElementById(runningApps).textContent stats.total_running || 0; document.getElementById(stoppedApps).textContent 0; // 需要从其他地方获取 // 计算总运行时间 let totalUptime 0; stats.apps.forEach(app { if (app.start_time) { const start new Date(app.start_time); const now new Date(); totalUptime (now - start) / 1000; // 转换为秒 } }); document.getElementById(totalUptime).textContent formatDuration(totalUptime); }) .catch(error { console.error(加载统计信息失败:, error); }); } // 刷新所有数据 function refreshAll() { const refreshBtn document.querySelector(.refresh-btn); refreshBtn.classList.add(loading); loadApps(); loadStats(); if (selectedAppId) { loadAppDetail(selectedAppId); loadHealthStatus(selectedAppId); loadLogs(currentLogType); } setTimeout(() { refreshBtn.classList.remove(loading); }, 1000); } // 计算运行时间 function calculateUptime(startTime) { const start new Date(startTime); const now new Date(); const duration (now - start) / 1000; // 转换为秒 return formatDuration(duration); } // 格式化持续时间 function formatDuration(seconds) { const days Math.floor(seconds / (24 * 3600)); const hours Math.floor((seconds % (24 * 3600)) / 3600); const minutes Math.floor((seconds % 3600) / 60); const secs Math.floor(seconds % 60); let result ; if (days 0) result ${days}天; if (hours 0) result ${hours}小时; if (minutes 0) result ${minutes}分钟; if (secs 0 || result ) result ${secs}秒; return result; } // 格式化时间 function formatTime(timeString) { const date new Date(timeString); return date.toLocaleString(zh-CN, { year: numeric, month: 2-digit, day: 2-digit, hour: 2-digit, minute: 2-digit, second: 2-digit }); } // 显示通知 function showNotification(message, type info) { const notification document.getElementById(notification); notification.textContent message; notification.className notification ${type} show; setTimeout(() { notification.classList.remove(show); }, 3000); } // 页面卸载时清除定时器 window.addEventListener(beforeunload, () { if (refreshInterval) { clearInterval(refreshInterval); } }); /script /body /htmlREADME.md# 多子项目协同中枢设计文档 ## 项目概述 本项目旨在建立一个多子项目协同中枢实现信息系统架构语言文档 抽象层与指令集架构程序文件 逻辑层的双架构融合设计。系统包含三层架构规则定义层、应用生成层和应用运行层通过物理层规约库文本将映射固化在本地数据中心。 ## 核心设计理念 ### 1. 忠实原文的备忘录仪表盘程序 备忘录仪表盘作为系统的核心组件负责 - 记录执行历史、结果和当前进度 - 作为协同看板和指挥棒 - 支持动态重构和灵活的模块化框架 ### 2. 多子项目协同设计 将备忘录仪表盘改造为多子项目协同的设计 - 备忘录设计层内部改造支持动态更新的组织结构 - 实现层对外接口改造建立灵活的模块化框架 - 同时支持做笔记变量约束和记日志目标绑定 - 仪表盘行为重组随人类行为而动重定位、重配置、重命名等 ### 3. 物理层规约 将映射固化在本地数据中心最小核心实现 - 通过设计不同的量词来控制不同的变量 - 将受控变量和本地可用的指令对应起来映射系统化 - 建立三个逻辑专项描述 - commands实现-细节 - requests设计-概要 - requirements需求 ## 三层架构设计 ### 规则定义层第一套 - **功能**定义生成应用的规则/标准 - **核心组件** - rule_lib.py规则处理核心库注册、查询、校验 - rule_define_app.py规则定义HTTP服务 - rule_define.html规则定义前端页面 - **端口**8001 - **数据存储**rule_db.json ### 应用生成层第二套 - **功能**基于规则自动生成可运行应用 - **核心组件** - generate_lib.py应用生成核心库代码生成、部署 - app_generate_app.py应用生成HTTP服务 - app_generate.html应用生成前端页面 - **端口**8002 - **数据存储**app_db.json ### 应用运行层第三套 - **功能**启动/停止/使用生成的应用 - **核心组件** - run_lib.py应用运行核心库启停、状态监控 - app_run_app.py应用运行HTTP服务 - app_run.html应用运行前端页面 - **端口**8003 - **数据存储**running_apps.json ## 技术栈 - **后端**纯原生Python无第三方依赖Python 3.6 - **前端**原生HTML/JS/CSS - **数据存储**JSON文件 - **通信**HTTP服务 - **部署**本地部署或云服务器部署 ## 项目结构 synergy_center_kit/ ├── rule_layer/ │ ├── lib/ │ │ └── rule_lib.py │ ├── app/ │ │ └── rule_define_app.py │ └── page/ │ └── rule_define.html ├── generate_layer/ │ ├── lib/ │ │ └── generate_lib.py │ ├── app/ │ │ └── app_generate_app.py │ └── page/ │ └── app_generate.html ├── run_layer/ │ ├── lib/ │ │ └── run_lib.py │ ├── app/ │ │ └── app_run_app.py │ └── page/ │ └── app_run.html ├── data/ │ ├── rule_db.json │ ├── app_db.json │ └── running_apps.json ├── start_all.sh ├── start_all.bat └── README.md ## 整体设计模板 根据讨论确定的内容整体设计包含三个核心部分 ### 1. 开发项目程序文件Creator - **面向目标**聚变-核聚变式工业实践革新 - **核心组件**笔记本工件库 域防腐层(ANIs) - **文档**DEVELOPMENT.md模板 - **对应**commands实现-细节 ### 2. 搭建开发环境Builder - **面向结果**突现-基因突变式科学创新 - **核心组件**面向对象的日志表工具箱 宏隐藏层(CLIs) - **脚本**start.sh 模板 - **对应**requests设计-概要 ### 3. 配套开发指南Checker - **面向过程**渐显-循序渐进式现象维新 - **核心组件**网络备忘录仪表盘 微隔离层(GUIs) - **文档**readme.txt模板 - **对应**requirements需求 ## 设计格式与样式 每一段由--引导的文字两侧三对文字每一对分别有确定的固定选项 - **前**格式 - 文档编辑图式概念创作 专文 - **后**样式 - 程序编程模式逻辑运作 专项 前面是具有固定结构的不同固定方法可以有很多版本演绎但每一次结构完整性要求相同且原始版本只有一个。这个原始版本是整体设计最先要交付的基础实现。 ## 后续扩展 系统设计支持后续扩展包括 - 连接云端资源组建局部网络 - 实现本地数据中心、局域网服务器和云端服务的协同 - 支持中蕴式AI组件、外挂式AI系统和内嵌式AI芯片 ## 开发计划 1. 实现三层架构的基础框架 2. 开发备忘录仪表盘核心功能 3. 实现多子项目协同机制 4. 完成物理层规约 5. 测试和优化系统性能 6. 文档完善和部署指南start_all.batecho off echo 启动多子项目协同中枢... echo :: 创建生成的应用目录 mkdir generated_apps :: 启动规则定义服务 echo 启动规则定义服务... start 规则定义服务 cmd /k cd rule_layer\app python rule_define_app.py :: 等待服务启动 timeout /t 2 nul :: 启动应用生成服务 echo 启动应用生成服务... start 应用生成服务 cmd /k cd generate_layer\app python app_generate_app.py :: 等待服务启动 timeout /t 2 nul :: 启动应用运行服务 echo ⚙️ 启动应用运行服务... start 应用运行服务 cmd /k cd run_layer\app python app_run_app.py echo echo 所有服务启动成功 echo. echo 访问地址 echo - 规则定义中心: http://localhost:8001 echo - 应用生成中心: http://localhost:8002 echo - 应用运行中心: http://localhost:8003 echo. echo 使用说明 echo 1. 首先在规则定义中心创建规则 echo 2. 然后在应用生成中心生成应用 echo 3. 最后在应用运行中心管理应用 echo. echo ⚠️ 注意关闭所有命令窗口可以停止服务 echo echo 服务状态 echo echo ✅ 规则定义服务: 运行中 echo ✅ 应用生成服务: 运行中 echo ✅ 应用运行服务: 运行中 echo echo 多子项目协同中枢已就绪 echo. echo 提示 echo - 关闭所有命令窗口可以停止服务 echo - 使用 stop_all.bat 脚本可以停止所有服务 echo. echo 开始使用吧 :: 等待用户按任意键继续 pausestart_all.sh#!/bin/bash # 启动所有服务的脚本 echo 启动多子项目协同中枢... echo # 创建生成的应用目录 mkdir -p generated_apps # 启动规则定义服务 echo 启动规则定义服务... cd rule_layer/app python3 rule_define_app.py RULE_PID$! cd ../.. echo ✅ 规则定义服务已启动 (PID: $RULE_PID) # 等待服务启动 sleep 2 # 启动应用生成服务 echo 启动应用生成服务... cd generate_layer/app python3 app_generate_app.py GENERATE_PID$! cd ../.. echo ✅ 应用生成服务已启动 (PID: $GENERATE_PID) # 等待服务启动 sleep 2 # 启动应用运行服务 echo ⚙️ 启动应用运行服务... cd run_layer/app python3 app_run_app.py RUN_PID$! cd ../.. echo ✅ 应用运行服务已启动 (PID: $RUN_PID) echo echo 所有服务启动成功 echo echo 访问地址 echo - 规则定义中心: http://localhost:8001 echo - 应用生成中心: http://localhost:8002 echo - 应用运行中心: http://localhost:8003 echo echo 使用说明 echo 1. 首先在规则定义中心创建规则 echo 2. 然后在应用生成中心生成应用 echo 3. 最后在应用运行中心管理应用 echo echo ⚠️ 注意按 CtrlC 可以停止所有服务 echo # 保存PID到文件 echo $RULE_PID .rule_pid echo $GENERATE_PID .generate_pid echo $RUN_PID .run_pid # 等待用户输入 echo 按任意键查看服务状态... read -n 1 # 显示服务状态 echo echo 服务状态 echo # 检查规则定义服务 if ps -p $RULE_PID /dev/null; then echo ✅ 规则定义服务: 运行中 (PID: $RULE_PID) else echo ❌ 规则定义服务: 已停止 fi # 检查应用生成服务 if ps -p $GENERATE_PID /dev/null; then echo ✅ 应用生成服务: 运行中 (PID: $GENERATE_PID) else echo ❌ 应用生成服务: 已停止 fi # 检查应用运行服务 if ps -p $RUN_PID /dev/null; then echo ✅ 应用运行服务: 运行中 (PID: $RUN_PID) else echo ❌ 应用运行服务: 已停止 fi echo echo 多子项目协同中枢已就绪 echo echo 提示 echo - 使用 kill $(cat .rule_pid .generate_pid .run_pid) 可以停止所有服务 echo - 使用 ./stop_all.sh 脚本可以优雅地停止所有服务 echo - 查看日志可以使用 tail -f rule_layer/app/*.log 等命令 echo echo 开始使用吧主页面