import os import json import sqlite3 import requests import random import time from flask import Flask, render_template, request, jsonify, g, send_from_directory from werkzeug.utils import secure_filename from werkzeug.exceptions import HTTPException app = Flask(__name__) # Config app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 # 16MB Limit app.config['UPLOAD_FOLDER'] = os.path.join(app.instance_path, 'uploads') DB_PATH = os.path.join(app.instance_path, 'material_mind.db') # API Configuration (SiliconFlow) SILICONFLOW_API_KEY = os.environ.get("SILICONFLOW_API_KEY", "sk-vimuseiptfbomzegyuvmebjzooncsqbyjtlddrfodzcdskgi") SILICONFLOW_API_URL = "https://api.siliconflow.cn/v1/chat/completions" # Ensure directories exist try: os.makedirs(app.instance_path, exist_ok=True) os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True) except OSError: pass # Database Helpers def get_db(): db = getattr(g, '_database', None) if db is None: db = g._database = sqlite3.connect(DB_PATH) db.row_factory = sqlite3.Row return db @app.teardown_appcontext def close_connection(exception): db = getattr(g, '_database', None) if db is not None: db.close() def init_db(): with app.app_context(): db = get_db() # Experiments Table db.execute(''' CREATE TABLE IF NOT EXISTS experiments ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, composition TEXT NOT NULL, -- JSON string properties TEXT, -- JSON string notes TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # Datasets Table (New) db.execute(''' CREATE TABLE IF NOT EXISTS datasets ( id INTEGER PRIMARY KEY AUTOINCREMENT, filename TEXT NOT NULL, filepath TEXT NOT NULL, description TEXT, uploaded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') db.commit() # Initialize DB init_db() # --- Global Error Handlers --- @app.errorhandler(404) def page_not_found(e): return render_template('index.html'), 200 # SPA fallback or error page @app.errorhandler(500) def internal_server_error(e): return jsonify(error="Internal Server Error", message=str(e)), 500 @app.errorhandler(413) def request_entity_too_large(e): return jsonify(error="File too large", message="File exceeds 16MB limit"), 413 # --- Routes --- @app.route('/') def index(): return render_template('index.html') @app.route('/api/experiments', methods=['GET']) def get_experiments(): db = get_db() cur = db.execute('SELECT * FROM experiments ORDER BY created_at DESC') rows = cur.fetchall() experiments = [] for row in rows: experiments.append({ 'id': row['id'], 'title': row['title'], 'composition': json.loads(row['composition']), 'properties': json.loads(row['properties']) if row['properties'] else {}, 'notes': row['notes'], 'created_at': row['created_at'] }) return jsonify(experiments) @app.route('/api/experiments', methods=['POST']) def create_experiment(): data = request.json title = data.get('title', 'Untitled Experiment') composition = json.dumps(data.get('composition', {})) properties = json.dumps(data.get('properties', {})) notes = data.get('notes', '') db = get_db() cur = db.execute( 'INSERT INTO experiments (title, composition, properties, notes) VALUES (?, ?, ?, ?)', (title, composition, properties, notes) ) db.commit() return jsonify({'id': cur.lastrowid, 'status': 'success'}) @app.route('/api/experiments/', methods=['DELETE']) def delete_experiment(experiment_id): db = get_db() db.execute('DELETE FROM experiments WHERE id = ?', (experiment_id,)) db.commit() return jsonify({'status': 'success'}) # --- Dataset/File Upload Routes --- @app.route('/api/datasets', methods=['GET']) def get_datasets(): db = get_db() cur = db.execute('SELECT * FROM datasets ORDER BY uploaded_at DESC') rows = cur.fetchall() datasets = [] for row in rows: datasets.append({ 'id': row['id'], 'filename': row['filename'], 'description': row['description'], 'uploaded_at': row['uploaded_at'] }) return jsonify(datasets) @app.route('/api/upload', methods=['POST']) def upload_file(): if 'file' not in request.files: return jsonify({'error': 'No file part'}), 400 file = request.files['file'] if file.filename == '': return jsonify({'error': 'No selected file'}), 400 if file: filename = secure_filename(file.filename) filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename) file.save(filepath) description = request.form.get('description', 'Uploaded Dataset') db = get_db() db.execute('INSERT INTO datasets (filename, filepath, description) VALUES (?, ?, ?)', (filename, filepath, description)) db.commit() return jsonify({'status': 'success', 'filename': filename}) # --- Simulation Logic --- @app.route('/api/simulate', methods=['POST']) def simulate_properties(): data = request.json composition = data.get('composition', {}) # Base Properties base_strength = 200 base_ductility = 50 base_cost = 10 # Contribution factors (Mock Database) factors = { 'Fe': {'strength': 2.0, 'ductility': 1.0, 'cost': 1.0, 'corrosion': 1.0}, 'C': {'strength': 12.0, 'ductility': -6.0, 'cost': 2.0, 'corrosion': -2.0}, 'Ni': {'strength': 4.0, 'ductility': 3.0, 'cost': 15.0, 'corrosion': 8.0}, 'Cr': {'strength': 5.0, 'ductility': 0.5, 'cost': 12.0, 'corrosion': 10.0}, 'Ti': {'strength': 9.0, 'ductility': -2.0, 'cost': 25.0, 'corrosion': 6.0}, 'Al': {'strength': 2.5, 'ductility': 0.0, 'cost': 5.0, 'corrosion': 4.0}, 'Cu': {'strength': 1.5, 'ductility': 2.0, 'cost': 8.0, 'corrosion': 3.0}, 'Mn': {'strength': 3.0, 'ductility': 1.5, 'cost': 3.0, 'corrosion': 1.0}, } total_strength = base_strength total_ductility = base_ductility total_cost = base_cost total_corrosion = 50 # Base score # Normalize composition total_percent = sum(float(v) for v in composition.values()) if total_percent == 0: total_percent = 1 for elem, amount in composition.items(): try: amount = float(amount) except ValueError: amount = 0 f = factors.get(elem, {'strength': 1, 'ductility': 0, 'cost': 1, 'corrosion': 0}) # Contribution Model total_strength += f['strength'] * amount * 0.6 total_ductility += f['ductility'] * amount * 0.4 total_cost += f['cost'] * amount * 0.1 total_corrosion += f['corrosion'] * amount * 0.5 # Apply some non-linear interactions (Mocking complex physics) # E.g., Cr + Ni synergy for corrosion cr = float(composition.get('Cr', 0)) ni = float(composition.get('Ni', 0)) if cr > 10 and ni > 5: total_corrosion *= 1.2 # Synergy bonus # Constraints total_strength = max(50, round(total_strength, 1)) total_ductility = max(0.1, round(total_ductility, 1)) total_cost = max(1, round(total_cost, 1)) total_corrosion = min(100, max(0, round(total_corrosion, 1))) return jsonify({ 'tensile_strength': total_strength, 'ductility': total_ductility, 'cost_index': total_cost, 'corrosion_resistance': total_corrosion, 'melting_point': 1500 - (float(composition.get('C', 0)) * 50) + (float(composition.get('W', 0)) * 20) # Fake }) # --- Chat & AI Logic --- @app.route('/api/chat', methods=['POST']) def chat(): data = request.json user_message = data.get('message', '') history = data.get('history', []) # System Prompt with specific instruction to return JSON for charts if needed system_prompt = { "role": "system", "content": ( "你是智材灵动（Material Mind）的AI助手，一位资深的材料科学家。" "请用专业、严谨但易懂的中文回答。" "如果你需要展示数据趋势或图表，请在回复的最后附加一个JSON代码块，" "格式为: ```json:chart { \"type\": \"bar|line|pie\", \"data\": { ... }, \"title\": \"...\" } ```。" "例如展示钢材强度对比：```json:chart { \"type\": \"bar\", \"title\": \"不同合金强度对比\", \"labels\": [\"合金A\", \"合金B\"], \"datasets\": [{ \"label\": \"强度(MPa)\", \"data\": [450, 600] }] } ```" ) } messages = [system_prompt] + history + [{"role": "user", "content": user_message}] headers = { "Authorization": f"Bearer {SILICONFLOW_API_KEY}", "Content-Type": "application/json" } payload = { "model": "Qwen/Qwen2.5-7B-Instruct", "messages": messages, "stream": False, "max_tokens": 1024 } try: response = requests.post(SILICONFLOW_API_URL, json=payload, headers=headers, timeout=30) response.raise_for_status() result = response.json() ai_content = result['choices'][0]['message']['content'] return jsonify({'response': ai_content}) except Exception as e: print(f"API Error: {e}") return mock_chat_response(user_message) def mock_chat_response(message): """ Mock Fallback that returns rich content (Markdown + JSON Charts) """ time.sleep(1) # Simulate network delay base_response = "**Mock Mode (云端连接中断)**: 正在使用本地应急知识库。\n\n" if "强度" in message or "strength" in message: return jsonify({'response': base_response + "关于材料强度，我们通常关注屈服强度和抗拉强度。添加碳(C)通常能显著提高钢的强度，但会降低延展性。\n\n" "以下是常见合金元素的强化效果对比：\n" "```json:chart\n" "{\n" " \"type\": \"bar\",\n" " \"title\": \"合金元素强化效果 (Mock Data)\",\n" " \"labels\": [\"碳 (C)\", \"锰 (Mn)\", \"硅 (Si)\", \"铬 (Cr)\"],\n" " \"datasets\": [{\n" " \"label\": \"强化系数\",\n" " \"data\": [12, 4, 3, 2],\n" " \"backgroundColor\": [\"#ef4444\", \"#3b82f6\", \"#10b981\", \"#f59e0b\"]\n" " }]\n" "}\n" "```" }) elif "腐蚀" in message or "corrosion" in message: return jsonify({'response': base_response + "提高耐腐蚀性的关键是形成致密的氧化膜。铬(Cr)是实现这一点的关键元素（如不锈钢需含Cr > 10.5%）。\n\n" "不锈钢耐腐蚀性随Cr含量变化趋势：\n" "```json:chart\n" "{\n" " \"type\": \"line\",\n" " \"title\": \"Cr含量与耐腐蚀性\",\n" " \"labels\": [\"0%\", \"5%\", \"10%\", \"15%\", \"20%\"],\n" " \"datasets\": [{\n" " \"label\": \"耐腐蚀指数\",\n" " \"data\": [10, 25, 80, 95, 98],\n" " \"borderColor\": \"#3b82f6\",\n" " \"fill\": true\n" " }]\n" "}\n" "```" }) else: return jsonify({'response': base_response + f"收到您的问题：“{message}”。\n" "作为一个材料科学助手，我可以帮您设计配方、预测性能或分析实验数据。\n" "尝试问我：“如何提高强度？”或者“不锈钢的配方是什么？”" }) if __name__ == '__main__': app.run(host='0.0.0.0', port=7860, debug=True)