from flask import Flask, jsonify, request, render_template_string from flask_sqlalchemy import SQLAlchemy from flask_migrate import Migrate from flask_jwt_extended import JWTManager, create_access_token, jwt_required, get_jwt_identity import uuid from datetime import datetime # Initialize Flask app app = Flask(__name__) app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///flowforge.db' app.config['JWT_SECRET_KEY'] = 'super-secret-2023' app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False # Initialize extensions db = SQLAlchemy(app) migrate = Migrate(app, db) jwt = JWTManager(app) # Database Models class User(db.Model): id = db.Column(db.String(36), primary_key=True, default=lambda: str(uuid.uuid4())) email = db.Column(db.String(120), unique=True, nullable=False) password = db.Column(db.String(120), nullable=False) tier = db.Column(db.String(20), default='free') # free, pro, enterprise class Project(db.Model): id = db.Column(db.String(36), primary_key=True, default=lambda: str(uuid.uuid4())) name = db.Column(db.String(120), nullable=False) owner_id = db.Column(db.String(36), db.ForeignKey('user.id'), nullable=False) deadline = db.Column(db.DateTime) class Task(db.Model): id = db.Column(db.String(36), primary_key=True, default=lambda: str(uuid.uuid4())) title = db.Column(db.String(120), nullable=False) project_id = db.Column(db.String(36), db.ForeignKey('project.id'), nullable=False) status = db.Column(db.String(20), default='todo') estimated_duration = db.Column(db.Integer) assigned_to = db.Column(db.String(36), db.ForeignKey('user.id')) created_at = db.Column(db.DateTime, default=datetime.utcnow) due_date = db.Column(db.DateTime) # AI Simulation Engine class AIService: @staticmethod def predict_duration(task_title): """Simulate AI duration prediction""" keyword_weights = {'design': 8, 'develop': 16, 'test': 4, 'review': 2, 'meeting': 1, 'document': 6} base_hours = 4 for kw, hours in keyword_weights.items(): if kw in task_title.lower(): return hours return base_hours @staticmethod def optimize_schedule(tasks): """Simulate AI schedule optimization""" return sorted(tasks, key=lambda x: x.get('estimated_duration', 0)) # Authentication Routes @app.route('/api/register', methods=['POST']) def register(): data = request.get_json() if User.query.filter_by(email=data['email']).first(): return jsonify({'error': 'Email already exists'}), 400 new_user = User( email=data['email'], password=data['password'], # In production: use bcrypt tier=data.get('tier', 'free') ) db.session.add(new_user) db.session.commit() return jsonify({'message': 'User created'}), 201 @app.route('/api/login', methods=['POST']) def login(): data = request.get_json() user = User.query.filter_by(email=data['email']).first() if not user or user.password != data['password']: return jsonify({'error': 'Invalid credentials'}), 401 access_token = create_access_token(identity=user.id) return jsonify(access_token=access_token, tier=user.tier), 200 # Project Management Routes @app.route('/api/projects', methods=['POST']) @jwt_required() def create_project(): current_user = get_jwt_identity() data = request.get_json() # Check project limit for free tier if User.query.get(current_user).tier == 'free' and \ Project.query.filter_by(owner_id=current_user).count() >= 3: return jsonify({'error': 'Free tier limited to 3 projects'}), 403 new_project = Project( name=data['name'], owner_id=current_user, deadline=datetime.fromisoformat(data['deadline']) if data.get('deadline') else None ) db.session.add(new_project) db.session.commit() return jsonify({'id': new_project.id, 'name': new_project.name}), 201 @app.route('/api/projects', methods=['GET']) @jwt_required() def get_projects(): current_user = get_jwt_identity() projects = Project.query.filter_by(owner_id=current_user).all() return jsonify([{'id': p.id, 'name': p.name} for p in projects]), 200 # Task Management Routes @app.route('/api/tasks', methods=['POST']) @jwt_required() def create_task(): current_user = get_jwt_identity() data = request.get_json() # AI prediction predicted_duration = AIService.predict_duration(data['title']) new_task = Task( title=data['title'], project_id=data['project_id'], estimated_duration=predicted_duration, due_date=datetime.fromisoformat(data['due_date']) if data.get('due_date') else None ) db.session.add(new_task) db.session.commit() return jsonify({ 'id': new_task.id, 'title': new_task.title, 'predicted_duration': predicted_duration }), 201 @app.route('/api/ai-schedule/', methods=['GET']) @jwt_required() def ai_schedule(project_id): tasks = Task.query.filter_by(project_id=project_id).all() if not tasks: return jsonify({'error': 'No tasks found'}), 404 optimized = AIService.optimize_schedule([ {'id': t.id, 'title': t.title, 'estimated_duration': t.estimated_duration} for t in tasks ]) return jsonify(optimized), 200 # Frontend Route @app.route('/') def index(): return render_template_string(''' FlowForge SaaS

FlowForge SaaS

AI-Powered Project Management

Login to Your Account

Don't have an account? Register

Create New Account

Already have an account? Login

Create New Project

Your Projects

Create New Task

Project Tasks

AI Schedule Optimization

''') # Initialize Database @app.before_first_request def initialize_database(): db.create_all() # Create sample user for testing if not User.query.first(): sample_user = User( email="test@flowforge.com", password="password123", tier="pro" ) db.session.add(sample_user) db.session.commit() # Run the application if __name__ == '__main__': app.run(debug=True)