mirror of
https://github.com/Pythagora-io/gpt-pilot.git
synced 2026-01-10 05:27:54 -05:00
Refactored saving files and file snapshot so they are connected
This commit is contained in:
@@ -289,8 +289,8 @@ def get_all_connected_steps(step, previous_step_field_name):
|
||||
prev_step = getattr(prev_step, previous_step_field_name)
|
||||
return connected_steps
|
||||
|
||||
def delete_all_steps_from_app(app):
|
||||
models = [DevelopmentSteps, CommandRuns, UserInputs]
|
||||
def delete_all_app_development_data(app):
|
||||
models = [DevelopmentSteps, CommandRuns, UserInputs, File, FileSnapshot]
|
||||
for model in models:
|
||||
model.delete().where(model.app == app).execute()
|
||||
|
||||
|
||||
@@ -2,14 +2,17 @@ from peewee import *
|
||||
|
||||
from database.models.components.base_models import BaseModel
|
||||
from database.models.development_steps import DevelopmentSteps
|
||||
from database.models.app import App
|
||||
from database.models.files import File
|
||||
|
||||
class FileSnapshot(BaseModel):
|
||||
app = ForeignKeyField(App, on_delete='CASCADE')
|
||||
development_step = ForeignKeyField(DevelopmentSteps, backref='files', on_delete='CASCADE')
|
||||
name = CharField()
|
||||
file = ForeignKeyField(File, on_delete='CASCADE', null=True)
|
||||
content = TextField()
|
||||
|
||||
class Meta:
|
||||
db_table = 'file_snapshot'
|
||||
indexes = (
|
||||
(('development_step', 'name'), True),
|
||||
(('development_step', 'file'), True),
|
||||
)
|
||||
@@ -5,10 +5,12 @@ from database.models.development_steps import DevelopmentSteps
|
||||
from database.models.app import App
|
||||
|
||||
class File(BaseModel):
|
||||
id = AutoField()
|
||||
app = ForeignKeyField(App, on_delete='CASCADE')
|
||||
name = CharField()
|
||||
path = CharField()
|
||||
description = TextField()
|
||||
full_path = CharField()
|
||||
description = TextField(null=True)
|
||||
|
||||
class Meta:
|
||||
indexes = (
|
||||
|
||||
@@ -3,9 +3,9 @@ import os
|
||||
from termcolor import colored
|
||||
from const.common import IGNORE_FOLDERS
|
||||
from database.models.app import App
|
||||
from database.database import get_app, delete_unconnected_steps_from, delete_all_steps_from_app
|
||||
from database.database import get_app, delete_unconnected_steps_from, delete_all_app_development_data
|
||||
from utils.questionary import styled_text
|
||||
from helpers.files import get_files_content, clear_directory
|
||||
from helpers.files import get_files_content, clear_directory, update_file
|
||||
from helpers.cli import build_directory_tree
|
||||
from helpers.agents.TechLead import TechLead
|
||||
from helpers.agents.Developer import Developer
|
||||
@@ -30,16 +30,6 @@ class Project:
|
||||
'last_command_run': None,
|
||||
'last_development_step': None,
|
||||
}
|
||||
if 'skip_until_dev_step' in args:
|
||||
self.skip_until_dev_step = args['skip_until_dev_step']
|
||||
if args['skip_until_dev_step'] == '0':
|
||||
delete_all_steps_from_app(args['app_id'])
|
||||
self.skip_steps = False
|
||||
else:
|
||||
self.skip_until_dev_step = None
|
||||
self.skip_steps = True
|
||||
self.skip_steps = False if ('skip_until_dev_step' in args and args['skip_until_dev_step'] == '0') else True
|
||||
|
||||
# TODO make flexible
|
||||
# self.root_path = get_parent_folder('euclid')
|
||||
self.root_path = ''
|
||||
@@ -72,6 +62,20 @@ class Project:
|
||||
self.tech_lead = TechLead(self)
|
||||
self.development_plan = self.tech_lead.create_development_plan()
|
||||
|
||||
# TODO move to constructor eventually
|
||||
if 'skip_until_dev_step' in self.args:
|
||||
self.skip_until_dev_step = self.args['skip_until_dev_step']
|
||||
if self.args['skip_until_dev_step'] == '0':
|
||||
clear_directory(self.root_path, IGNORE_FOLDERS)
|
||||
delete_all_app_development_data(self.args['app_id'])
|
||||
self.skip_steps = False
|
||||
else:
|
||||
self.skip_until_dev_step = None
|
||||
self.skip_steps = True
|
||||
|
||||
self.skip_steps = False if ('skip_until_dev_step' in self.args and self.args['skip_until_dev_step'] == '0') else True
|
||||
# TODO END
|
||||
|
||||
self.developer = Developer(self)
|
||||
self.developer.set_up_environment();
|
||||
|
||||
@@ -79,21 +83,27 @@ class Project:
|
||||
|
||||
def get_directory_tree(self, with_descriptions=False):
|
||||
files = {}
|
||||
if with_descriptions:
|
||||
if with_descriptions and False:
|
||||
files = File.select().where(File.app_id == self.args['app_id'])
|
||||
files = {snapshot.name: snapshot for snapshot in files}
|
||||
return build_directory_tree(self.root_path + '/', ignore=IGNORE_FOLDERS, files=files, add_descriptions=True)
|
||||
return build_directory_tree(self.root_path + '/', ignore=IGNORE_FOLDERS, files=files, add_descriptions=False)
|
||||
|
||||
def get_test_directory_tree(self):
|
||||
# TODO remove hardcoded path
|
||||
return build_directory_tree(self.root_path + '/tests', ignore=IGNORE_FOLDERS)
|
||||
|
||||
def get_all_coded_files(self):
|
||||
files = File.select().where(File.app_id == self.args['app_id'])
|
||||
files = self.get_files([file.path + file.name for file in files])
|
||||
return files
|
||||
|
||||
def get_files(self, files):
|
||||
files_with_content = []
|
||||
for file in files:
|
||||
# TODO this is a hack, fix it
|
||||
try:
|
||||
file_content = open(self.get_full_file_path('', file), 'r').read()
|
||||
relative_path, full_path = self.get_full_file_path('', file)
|
||||
file_content = open(full_path, 'r').read()
|
||||
except:
|
||||
file_content = ''
|
||||
|
||||
@@ -103,22 +113,48 @@ class Project:
|
||||
})
|
||||
return files_with_content
|
||||
|
||||
def save_file(self, data):
|
||||
data['path'], data['full_path'] = self.get_full_file_path(data['path'], data['name'])
|
||||
update_file(data['full_path'], data['content'])
|
||||
|
||||
file_in_db, created = File.get_or_create(
|
||||
app=self.app,
|
||||
name=data['name'],
|
||||
path=data['path'],
|
||||
full_path=data['full_path'],
|
||||
)
|
||||
|
||||
def get_full_file_path(self, file_path, file_name):
|
||||
file_path = file_path.replace('./', '', 1).rstrip(file_name)
|
||||
|
||||
if not file_path.endswith('/'):
|
||||
file_path = file_path + '/'
|
||||
|
||||
if file_name.startswith('/'):
|
||||
file_name = file_name[1:]
|
||||
|
||||
if not file_path.startswith('/'):
|
||||
file_path = '/' + file_path
|
||||
return self.root_path + file_path + file_name
|
||||
|
||||
return (file_path, self.root_path + file_path + file_name)
|
||||
|
||||
def save_files_snapshot(self, development_step_id):
|
||||
files = get_files_content(self.root_path, ignore=IGNORE_FOLDERS)
|
||||
development_step, created = DevelopmentSteps.get_or_create(id=development_step_id)
|
||||
|
||||
for file in files:
|
||||
file_snapshot, created = FileSnapshot.get_or_create(
|
||||
development_step=development_step,
|
||||
# TODO this can be optimized so we don't go to the db each time
|
||||
file_in_db, created = File.get_or_create(
|
||||
app=self.app,
|
||||
name=file['name'],
|
||||
path=file['path'],
|
||||
full_path=file['full_path'],
|
||||
)
|
||||
|
||||
file_snapshot, created = FileSnapshot.get_or_create(
|
||||
app=self.app,
|
||||
development_step=development_step,
|
||||
file=file_in_db,
|
||||
defaults={'content': file.get('content', '')}
|
||||
)
|
||||
file_snapshot.content = content = file['content']
|
||||
@@ -130,7 +166,7 @@ class Project:
|
||||
|
||||
clear_directory(self.root_path, IGNORE_FOLDERS)
|
||||
for file_snapshot in file_snapshots:
|
||||
full_path = self.root_path + '/' + file_snapshot.name
|
||||
full_path = self.root_path + file_snapshot.file.path + '/' + file_snapshot.file.name
|
||||
# Ensure directory exists
|
||||
os.makedirs(os.path.dirname(full_path), exist_ok=True)
|
||||
|
||||
|
||||
@@ -28,16 +28,6 @@ class CodeMonkey(Agent):
|
||||
}, IMPLEMENT_CHANGES)
|
||||
|
||||
for file_data in changes:
|
||||
file_data['full_path'] = self.project.get_full_file_path(file_data['path'], file_data['name'])
|
||||
|
||||
if file_data['description'] != '':
|
||||
(File.insert(app=self.project.app, path=file_data['path'], name=file_data['name'], description=file_data['description'])
|
||||
.on_conflict(
|
||||
conflict_target=[File.app, File.name, File.path],
|
||||
preserve=[],
|
||||
update={'description': file_data['description']})
|
||||
.execute())
|
||||
|
||||
update_file(file_data['full_path'], file_data['content'])
|
||||
self.project.save_file(file_data)
|
||||
|
||||
return convo
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import json
|
||||
import uuid
|
||||
from termcolor import colored
|
||||
from helpers.files import update_file
|
||||
from utils.utils import step_already_finished
|
||||
from helpers.agents.CodeMonkey import CodeMonkey
|
||||
from logger.logger import logger
|
||||
@@ -98,7 +99,7 @@ class Developer(Agent):
|
||||
if step and not execute_step(self.project.args['step'], self.project.current_step):
|
||||
step_already_finished(self.project.args, step)
|
||||
return
|
||||
|
||||
|
||||
# ENVIRONMENT SETUP
|
||||
print(colored(f"Setting up the environment...\n", "green"))
|
||||
logger.info(f"Setting up the environment...")
|
||||
|
||||
@@ -28,9 +28,12 @@ def get_files_content(directory, ignore=[]):
|
||||
with open(path, 'r', encoding='utf-8', errors='ignore') as f:
|
||||
file_content = f.read()
|
||||
|
||||
file_name = path.replace(directory + '/', '')
|
||||
return_array.append({
|
||||
'name': path.replace(directory + '/', ''),
|
||||
'content': file_content
|
||||
'name': file_name,
|
||||
'path': '/' + file.replace(file_name, ''),
|
||||
'content': file_content,
|
||||
'full_path': path,
|
||||
})
|
||||
|
||||
return return_array
|
||||
|
||||
Reference in New Issue
Block a user