diff --git a/backend/app/api/routes/scripts.py b/backend/app/api/routes/scripts.py
index a8df816..ee4a373 100644
--- a/backend/app/api/routes/scripts.py
+++ b/backend/app/api/routes/scripts.py
@@ -8,7 +8,7 @@
from typing import Dict, Any, Optional, List
import uuid
-from ...services.parser.renpy_parser import RenPyParser, ChoiceNode, ChoiceNodeType
+# Removed parser import
from ...services.database import DatabaseService
from ...models.exceptions import ResourceNotFoundException, DatabaseException
from ...services.websocket import connection_manager
@@ -23,24 +23,7 @@
# Initialize services
db_service = DatabaseService()
-parser = RenPyParser()
-
-# Helper functions
-def node_to_dict(node: ChoiceNode) -> Dict[str, Any]:
- """Convert a ChoiceNode to a dictionary with line references for JSON serialization."""
- result = {
- "id": str(id(node)), # Generate a unique ID using the object's memory address
- "node_type": node.node_type.value if hasattr(node.node_type, "value") else str(node.node_type),
- "label_name": node.label_name,
- "start_line": node.start_line,
- "end_line": node.end_line,
- "children": [node_to_dict(child) for child in node.children]
- }
-
- if hasattr(node, "false_branch") and node.false_branch:
- result["false_branch"] = [node_to_dict(opt) for opt in node.false_branch]
-
- return result
+# parser = RenPyParser() # Removed
# Routes
@scripts_router.post("/parse", response_model=Dict[str, Any])
@@ -51,101 +34,44 @@ async def parse_script(
current_user: Dict[str, Any] = Depends(get_current_user)
) -> Dict[str, Any]:
"""
- Parse a RenPy script file and return its tree structure with line references.
-
- Args:
- file: The uploaded RenPy script file
- project_id: ID of the project to associate the script with
-
- Returns:
- JSON representation of the parsed script tree with line references
+ Accepts a script file and returns its content for client-side parsing.
+ Legacy name "parse" kept for compatibility, but now it acts as an upload/read endpoint.
"""
try:
- # current_user now injected via Depends
- if not file.filename:
- raise HTTPException(status_code=400, detail="Filename is required")
-
- if not file.filename.lower().endswith(".rpy"):
- raise HTTPException(status_code=400, detail="Invalid file type. Only .rpy files are allowed.")
-
- # Read file content
- content = await file.read()
- file_size = len(content)
-
- if file_size > 1024 * 1024: # 1MB limit
- raise HTTPException(status_code=400, detail="File too large. Maximum size is 1MB.")
-
- # Verify user has access to the specified project
+ # Validate user has access to the project
user_projects = db_service.get_user_projects(current_user["id"])
has_access = any(p["id"] == project_id for p in user_projects)
if not has_access:
- raise HTTPException(status_code=403, detail="Access denied to the specified project")
-
- # Parse the content to check validity
- temp_dir = Path(tempfile.gettempdir()) / "renpy_editor" / str(uuid.uuid4())
- temp_dir.mkdir(parents=True, exist_ok=True)
-
- temp_file = temp_dir / file.filename
- with open(temp_file, "wb") as f:
- f.write(content)
-
- # Parse the file to build the tree
- parsed_tree = await parser.parse_async(str(temp_file))
+ raise HTTPException(status_code=403, detail="Access denied to this project")
+
+ content = await file.read()
+ content_str = content.decode('utf-8')
- # Check if script with this filename already exists in the project
- existing_script = db_service.get_script_by_filename(project_id, file.filename)
+ # Save to database (create new script entry)
+ filename = file.filename or "uploaded.rpy"
+ script_id = db_service.create_script(project_id, filename, content_str, current_user["id"])
- decoded_content = content.decode('utf-8')
-
- if existing_script:
- # Update existing script
- script_id = existing_script["id"]
- db_service.update_script(
- script_id=script_id,
- content=decoded_content,
- user_id=current_user["id"]
- )
- else:
- # Save to database
- script_id = db_service.save_script(
- project_id=project_id,
- filename=file.filename,
- content=decoded_content,
- user_id=current_user["id"]
- )
-
- # Clean up temp file
- background_tasks.add_task(shutil.rmtree, temp_dir, ignore_errors=True)
-
- # Return result
- result = {
+ return {
"script_id": script_id,
- "filename": file.filename,
- "tree": node_to_dict(parsed_tree)
+ "filename": filename,
+ "content": content_str,
+ # "tree": ... # No longer returned
}
-
- return result
+ except HTTPException:
+ raise
except Exception as e:
- raise HTTPException(status_code=500, detail=f"Error parsing script: {str(e)}")
+ raise HTTPException(status_code=500, detail=f"Error processing script: {str(e)}")
-@scripts_router.get("/node-content/{script_id}", response_model=Dict[str, Any])
+@scripts_router.get("/node-content/{script_id}", response_model=Dict[str, str])
async def get_node_content(
- script_id: str,
- start_line: int,
+ script_id: str,
+ start_line: int,
end_line: int,
token: str = Depends(oauth2_scheme)
-) -> Dict[str, Any]:
+) -> Dict[str, str]:
"""
- Get the content of a specific node by line range.
-
- Args:
- script_id: The ID of the script
- start_line: Starting line number (0-indexed)
- end_line: Ending line number (0-indexed)
-
- Returns:
- Node content
+ Get the raw content of a specific node by line numbers.
"""
try:
current_user = await get_current_user(token)
@@ -155,7 +81,7 @@ async def get_node_content(
if not script:
raise ResourceNotFoundException("Script", script_id)
- # Validate user has access to the script's project
+ # Validate user has access
project_id = script["project_id"]
user_projects = db_service.get_user_projects(current_user["id"])
has_access = any(p["id"] == project_id for p in user_projects)
@@ -165,131 +91,102 @@ async def get_node_content(
content_lines = script["content_lines"]
- # Validate line range
+ # Validate lines
if start_line < 0 or end_line >= len(content_lines) or start_line > end_line:
raise HTTPException(status_code=400, detail="Invalid line range")
+
+ node_content = "\n".join(content_lines[start_line:end_line+1])
- # Extract the content
- node_content = content_lines[start_line:end_line + 1]
-
- return {
- "content": "\n".join(node_content),
- "start_line": start_line,
- "end_line": end_line
- }
+ return {"content": node_content}
except ResourceNotFoundException as e:
raise HTTPException(status_code=404, detail=str(e))
except HTTPException:
raise
except Exception as e:
- raise HTTPException(status_code=500, detail=f"Error retrieving node content: {str(e)}")
+ raise HTTPException(status_code=500, detail=f"Error getting node content: {str(e)}")
@scripts_router.post("/update-node/{script_id}", response_model=Dict[str, Any])
async def update_node_content(
- script_id: str,
- start_line: int,
- end_line: int,
+ script_id: str,
+ start_line: int,
+ end_line: int,
content: str = Body(..., embed=True),
token: str = Depends(oauth2_scheme)
) -> Dict[str, Any]:
"""
- Update the content of a specific node.
-
- Args:
- script_id: The ID of the script
- start_line: Starting line number (0-indexed)
- end_line: Ending line number (0-indexed)
- content: New content for the node
-
- Returns:
- Updated line range
+ Update the content of a node.
"""
try:
current_user = await get_current_user(token)
- # Get script from database
+ # Get script
script = db_service.get_script(script_id)
if not script:
raise ResourceNotFoundException("Script", script_id)
-
- # Validate user has access to the script's project
+
+ # Validate access
project_id = script["project_id"]
user_projects = db_service.get_user_projects(current_user["id"])
has_access = any(p["id"] == project_id for p in user_projects)
if not has_access:
raise HTTPException(status_code=403, detail="Access denied to this script")
-
- # Get current content
+
content_lines = script["content_lines"]
- # Validate line range
+ # Validate lines
if start_line < 0 or end_line >= len(content_lines) or start_line > end_line:
raise HTTPException(status_code=400, detail="Invalid line range")
-
- # Calculate line differences
- old_line_count = end_line - start_line + 1
- new_content_lines = content.splitlines()
- new_line_count = len(new_content_lines)
- line_diff = new_line_count - old_line_count
-
- # Update the content
- content_lines[start_line:end_line+1] = new_content_lines
+
+ # Replace content
+ new_lines = content.splitlines()
+ content_lines[start_line:end_line+1] = new_lines
new_content = "\n".join(content_lines)
- # Save changes to database
+ # Save
db_service.update_script(script_id, new_content, current_user["id"])
- # Parse updated script and broadcast new structure to collaborators
- parsed_tree = parser.parse_text(new_content)
- await connection_manager.broadcast_structure_update(
- script_id, node_to_dict(parsed_tree)
+ # No re-parsing here. Client should handle update or reload.
+ # Broadcast raw update notification
+ # The client will likely need to re-fetch or re-parse locally if the structure changed.
+ # But for text content update, maybe structure didn't change?
+ # Ideally we broadcast "content_updated" and clients decide.
+
+ await connection_manager.broadcast_to_script(
+ script_id,
+ {
+ "type": "content_updated", # Generic update
+ "user_id": current_user["id"],
+ "timestamp": "now"
+ }
)
- # Calculate new end line
- new_end_line = start_line + new_line_count - 1
-
- return {
- "start_line": start_line,
- "end_line": new_end_line,
- "line_diff": line_diff
- }
+ return {"success": True, "message": "Node updated successfully"}
except ResourceNotFoundException as e:
raise HTTPException(status_code=404, detail=str(e))
except HTTPException:
raise
except Exception as e:
- raise HTTPException(status_code=500, detail=f"Error updating node content: {str(e)}")
+ raise HTTPException(status_code=500, detail=f"Error updating node: {str(e)}")
@scripts_router.post("/insert-node/{script_id}", response_model=Dict[str, Any])
async def insert_node(
- script_id: str,
- insertion_line: int,
- content: str = Body(..., embed=True),
- node_type: str = Body(..., embed=True),
+ script_id: str,
+ insertion_line: int,
+ node_type: str = Body(...),
+ content: str = Body(...),
token: str = Depends(oauth2_scheme)
) -> Dict[str, Any]:
"""
- Insert a new node at the specified position.
-
- Args:
- script_id: The ID of the script
- insertion_line: The line where to insert the new node
- content: The content of the new node
- node_type: The type of node to insert
-
- Returns:
- Information about the inserted node
+ Insert a new node at a specific line.
"""
try:
current_user = await get_current_user(token)
- # Get script from database
script = db_service.get_script(script_id)
if not script:
raise ResourceNotFoundException("Script", script_id)
- # Validate user has access to the script's project
project_id = script["project_id"]
user_projects = db_service.get_user_projects(current_user["id"])
has_access = any(p["id"] == project_id for p in user_projects)
@@ -297,48 +194,36 @@ async def insert_node(
if not has_access:
raise HTTPException(status_code=403, detail="Access denied to this script")
- # Get current content
content_lines = script["content_lines"]
- # Validate insertion line
if insertion_line < 0 or insertion_line > len(content_lines):
raise HTTPException(status_code=400, detail="Invalid insertion line")
- # Parse the new content
new_content_lines = content.splitlines()
-
- # Insert the new lines
content_lines[insertion_line:insertion_line] = new_content_lines
new_content = "\n".join(content_lines)
- # Save changes to database
db_service.update_script(script_id, new_content, current_user["id"])
- # Re-parse the entire script to update the tree
- # Create temp file for parsing
- temp_dir = Path(tempfile.gettempdir()) / "renpy_editor" / str(uuid.uuid4())
- temp_dir.mkdir(parents=True, exist_ok=True)
- temp_file = temp_dir / "temp_script.rpy"
+ # NO PARSING.
- with open(temp_file, "w", encoding="utf-8") as f:
- f.write(new_content)
-
- # Parse the updated file
- parsed_tree = await parser.parse_async(str(temp_file))
-
- # Clean up temp file
- shutil.rmtree(temp_dir, ignore_errors=True)
-
- # Broadcast updated structure to other clients
- await connection_manager.broadcast_structure_update(
- script_id, node_to_dict(parsed_tree)
+ # Broadcast insert
+ await connection_manager.broadcast_to_script(
+ script_id,
+ {
+ "type": "content_inserted",
+ "line": insertion_line,
+ "count": len(new_content_lines),
+ "user_id": current_user["id"]
+ }
)
return {
"start_line": insertion_line,
"end_line": insertion_line + len(new_content_lines) - 1,
"line_count": len(new_content_lines),
- "tree": node_to_dict(parsed_tree)
+ "content": content
+ # "tree": ... # Removed
}
except ResourceNotFoundException as e:
raise HTTPException(status_code=404, detail=str(e))
@@ -352,24 +237,13 @@ async def download_script(
script_id: str,
token: str = Depends(oauth2_scheme)
) -> JSONResponse:
- """
- Download the current version of the script.
-
- Args:
- script_id: The ID of the script
-
- Returns:
- The script file content
- """
try:
current_user = await get_current_user(token)
- # Get script from database
script = db_service.get_script(script_id)
if not script:
raise ResourceNotFoundException("Script", script_id)
- # Validate user has access to the script's project
project_id = script["project_id"]
user_projects = db_service.get_user_projects(current_user["id"])
has_access = any(p["id"] == project_id for p in user_projects)
@@ -393,24 +267,13 @@ async def delete_script(
script_id: str,
token: str = Depends(oauth2_scheme)
) -> Dict[str, str]:
- """
- Delete a script.
-
- Args:
- script_id: The ID of the script to delete
-
- Returns:
- Confirmation message
- """
try:
current_user = await get_current_user(token)
- # Get script from database
script = db_service.get_script(script_id)
if not script:
raise ResourceNotFoundException("Script", script_id)
- # Validate user has access to the script's project
project_id = script["project_id"]
user_projects = db_service.get_user_projects(current_user["id"])
has_access = any(p["id"] == project_id and p.get("role") in ["Owner", "Editor"]
@@ -419,7 +282,6 @@ async def delete_script(
if not has_access:
raise HTTPException(status_code=403, detail="Permission denied to delete this script")
- # Delete from database
deleted = db_service.delete_script(script_id)
if not deleted:
raise HTTPException(status_code=500, detail="Failed to delete script")
@@ -437,26 +299,15 @@ async def get_project_scripts(
project_id: str,
token: str = Depends(oauth2_scheme)
) -> List[Dict[str, Any]]:
- """
- Get all scripts for a project.
-
- Args:
- project_id: The ID of the project
-
- Returns:
- List of scripts in the project
- """
try:
current_user = await get_current_user(token)
- # Validate user has access to the project
user_projects = db_service.get_user_projects(current_user["id"])
has_access = any(p["id"] == project_id for p in user_projects)
if not has_access:
raise HTTPException(status_code=403, detail="Access denied to this project")
- # Get scripts from database
scripts = db_service.get_project_scripts(project_id)
return scripts
except HTTPException:
@@ -471,21 +322,9 @@ async def search_scripts(
limit: int = 20,
token: str = Depends(oauth2_scheme)
) -> List[Dict[str, Any]]:
- """
- Search scripts by content or filename.
-
- Args:
- query: Search query
- project_id: Optional project ID to filter by
- limit: Maximum number of results
-
- Returns:
- List of matching scripts
- """
try:
current_user = await get_current_user(token)
- # If project ID specified, validate access
if project_id:
user_projects = db_service.get_user_projects(current_user["id"])
has_access = any(p["id"] == project_id for p in user_projects)
@@ -493,7 +332,6 @@ async def search_scripts(
if not has_access:
raise HTTPException(status_code=403, detail="Access denied to this project")
- # Search scripts
results = db_service.search_scripts(
project_id=project_id,
query=query,
@@ -512,21 +350,13 @@ async def load_existing_script(
current_user: Dict[str, Any] = Depends(get_current_user)
) -> Dict[str, Any]:
"""
- Load an existing script and return its parsed tree structure.
-
- Args:
- script_id: The ID of the script to load
-
- Returns:
- JSON representation of the script with parsed tree
+ Load an existing script. Returns only content, parsing is done client-side.
"""
try:
- # Get script from database
script = db_service.get_script(script_id)
if not script:
raise ResourceNotFoundException("Script", script_id)
- # Validate user has access to the script's project
project_id = script["project_id"]
user_projects = db_service.get_user_projects(current_user["id"])
has_access = any(p["id"] == project_id for p in user_projects)
@@ -534,30 +364,13 @@ async def load_existing_script(
if not has_access:
raise HTTPException(status_code=403, detail="Access denied to this script")
- # Parse the script content to build tree
- temp_dir = Path(tempfile.gettempdir()) / "renpy_editor" / str(uuid.uuid4())
- temp_dir.mkdir(parents=True, exist_ok=True)
-
- temp_file = temp_dir / script["filename"]
- with open(temp_file, "w", encoding='utf-8') as f:
- f.write(script["content"])
-
- try:
- # Parse the file to build the tree
- parsed_tree = await parser.parse_async(str(temp_file))
-
- # Return result
- result = {
- "script_id": script_id,
- "filename": script["filename"],
- "tree": node_to_dict(parsed_tree)
- }
-
- return result
- finally:
- # Clean up temp file
- import shutil
- shutil.rmtree(temp_dir, ignore_errors=True)
+ # Return content without parsing
+ return {
+ "script_id": script_id,
+ "filename": script["filename"],
+ "content": script["content"]
+ # "tree": ... # Removed
+ }
except ResourceNotFoundException as e:
raise HTTPException(status_code=404, detail=str(e))
@@ -565,5 +378,3 @@ async def load_existing_script(
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error loading script: {str(e)}")
-
-# TODO: Add endpoints for version history retrieval - #issue/129
diff --git a/backend/app/services/database.py b/backend/app/services/database.py
index 4444a51..5157410 100644
--- a/backend/app/services/database.py
+++ b/backend/app/services/database.py
@@ -199,385 +199,357 @@ def _initialize_database(self):
logger.error(error_msg)
raise FileNotFoundError(error_msg)
+ # Initialize database with schema
with sqlite3.connect(self.db_path) as conn:
- # Set foreign keys pragma
- conn.execute("PRAGMA foreign_keys = ON")
-
- # Read and execute the schema SQL
- logger.debug("Executing schema script")
conn.executescript(schema_content)
- conn.commit()
-
- # Verify tables were created
- cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
- tables = [row[0] for row in cursor.fetchall()]
- logger.info(f"Tables created in database: {', '.join(tables)}")
-
- # Verify required tables exist
- required_tables = ["users", "projects", "scripts", "roles", "sessions", "participants", "node_locks"]
- missing_tables = [table for table in required_tables if table not in tables]
- if missing_tables:
- raise RuntimeError(f"Failed to create required tables: {', '.join(missing_tables)}")
+ logger.info("Database initialized successfully.")
- logger.info(f"Database initialized successfully at {self.db_path}")
except Exception as e:
logger.error(f"Database initialization failed: {str(e)}")
raise
-
+
def _get_connection(self):
- """Get a new database connection with proper settings."""
- # Removed connection caching to prevent threading issues with TestClient
- connection = sqlite3.connect(self.db_path, isolation_level=None, check_same_thread=False) # Allow cross-thread usage for FastAPI TestClient context
- connection.row_factory = sqlite3.Row # Return rows as dictionaries
- connection.execute("PRAGMA foreign_keys = ON") # Ensure foreign keys are enabled for each connection
- return connection
-
- def close(self):
- """Explicitly close any open database connections (if any were cached - now deprecated)."""
- # Connection caching is removed, so this method might be less critical,
- # but kept for potential future use or explicit cleanup needs.
+ """Get a database connection."""
+ conn = sqlite3.connect(self.db_path)
+ conn.row_factory = sqlite3.Row
+ return conn
+
+ # User methods
+ def get_user_by_username(self, username: str) -> Optional[Dict[str, Any]]:
+ """Get user by username."""
try:
- # Clear the script cache
- if hasattr(self, 'script_cache'):
- self.script_cache.clear()
+ with self._get_connection() as conn:
+ cursor = conn.execute(
+ "SELECT * FROM users WHERE username = ?",
+ (username,)
+ )
+ row = cursor.fetchone()
+ if row:
+ return dict(row)
+ return None
except Exception as e:
- print(f"Error during database service cleanup: {e}")
-
- def __del__(self):
- """Ensure cleanup on object garbage collection."""
- # No connection to close here anymore due to removal of caching
- pass
-
- def create_project(self, name: str, owner_id: str, description: str = None) -> str:
- """Create a new project and return its ID."""
- project_id = str(uuid.uuid4())
+ logger.error(f"Failed to get user by username: {str(e)}")
+ raise
+
+ def get_user_by_email(self, email: str) -> Optional[Dict[str, Any]]:
+ """Get user by email."""
try:
- # Handle None description by setting it to an empty string
- description = description or ""
- with sqlite3.connect(self.db_path) as conn:
+ with self._get_connection() as conn:
+ cursor = conn.execute(
+ "SELECT * FROM users WHERE email = ?",
+ (email,)
+ )
+ row = cursor.fetchone()
+ if row:
+ return dict(row)
+ return None
+ except Exception as e:
+ logger.error(f"Failed to get user by email: {str(e)}")
+ raise
+
+ def create_user(self, username: str, email: str, password_hash: str) -> str:
+ """Create a new user."""
+ user_id = str(uuid.uuid4())
+ try:
+ with self._get_connection() as conn:
conn.execute(
- 'INSERT INTO projects (id, name, description, owner_id) VALUES (?, ?, ?, ?)',
- (project_id, name, description, owner_id)
+ "INSERT INTO users (id, username, email, password_hash) VALUES (?, ?, ?, ?)",
+ (user_id, username, email, password_hash)
)
- logger.info(f"Created project {name} with ID {project_id}")
- return project_id
+ return user_id
+ except sqlite3.IntegrityError:
+ # Check if it was username or email constraint
+ existing_username = self.get_user_by_username(username)
+ if existing_username:
+ raise ValueError("Username already exists")
+ raise ValueError("Email already exists")
except Exception as e:
- logger.error(f"Failed to create project: {str(e)}")
+ logger.error(f"Failed to create user: {str(e)}")
raise
-
- def save_script(self, project_id: str, filename: str, content: str, user_id: Optional[str] = None) -> str:
- """Save a script to the database and return its ID."""
- script_id = str(uuid.uuid4())
+
+ # Project methods
+ def create_project(self, name: str, description: str, owner_id: str) -> str:
+ """Create a new project."""
+ project_id = str(uuid.uuid4())
try:
- with sqlite3.connect(self.db_path) as conn:
- conn.execute('BEGIN TRANSACTION')
- try:
- # Insert the script
- conn.execute(
- 'INSERT INTO scripts (id, project_id, filename, content, last_edited_by) VALUES (?, ?, ?, ?, ?)',
- (script_id, project_id, filename, content, user_id)
- )
-
- # Create initial version if user_id is provided
- if user_id:
- version_id = str(uuid.uuid4())
- conn.execute(
- 'INSERT INTO versions (id, script_id, content, created_by) VALUES (?, ?, ?, ?)',
- (version_id, script_id, content, user_id)
- )
-
- conn.execute('COMMIT')
- # Update cache
- self.script_cache.set(script_id, {
- "id": script_id,
- "project_id": project_id,
- "filename": filename,
- "content": content,
- "content_lines": content.splitlines(),
- "last_edited_by": user_id
- })
- logger.info(f"Saved script {filename} with ID {script_id}")
- return script_id
- except Exception as e:
- conn.execute('ROLLBACK')
- raise
+ with self._get_connection() as conn:
+ conn.execute(
+ "INSERT INTO projects (id, name, description, owner_id) VALUES (?, ?, ?, ?)",
+ (project_id, name, description, owner_id)
+ )
+ # Add owner access
+ conn.execute(
+ "INSERT INTO project_access (project_id, user_id, role_id) VALUES (?, ?, ?)",
+ (project_id, owner_id, 'role_owner')
+ )
+ return project_id
except Exception as e:
- logger.error(f"Failed to save script: {str(e)}")
+ logger.error(f"Failed to create project: {str(e)}")
raise
-
- def update_script(self, script_id: str, content: str, user_id: str) -> None:
- """Update script content and create a version entry."""
- # Invalidate cache for this script
- self.script_cache.delete(script_id)
+
+ def get_user_projects(self, user_id: str) -> List[Dict[str, Any]]:
+ """Get all projects accessible to a user."""
try:
- with sqlite3.connect(self.db_path) as conn:
- # Begin a transaction
- conn.execute('BEGIN TRANSACTION')
- try:
- conn.execute(
- 'UPDATE scripts SET content = ?, updated_at = CURRENT_TIMESTAMP, last_edited_by = ? WHERE id = ?',
- (content, user_id, script_id)
- )
- conn.execute(
- 'INSERT INTO versions (id, script_id, content, created_by) VALUES (?, ?, ?, ?)',
- (str(uuid.uuid4()), script_id, content, user_id)
- )
- # Commit the transaction
- conn.execute('COMMIT')
- # Update cache if exists
- cached_script = self.script_cache.get(script_id)
- if cached_script:
- cached_script.update({
- "content": content,
- "content_lines": content.splitlines(),
- "last_edited_by": user_id
- })
- self.script_cache.set(script_id, cached_script)
- logger.info(f"Updated script {script_id} by user {user_id}")
- except Exception as e:
- # Rollback in case of error
- conn.execute('ROLLBACK')
- logger.error(f"Failed to update script (transaction rolled back): {str(e)}")
- raise
+ with self._get_connection() as conn:
+ cursor = conn.execute(
+ '''
+ SELECT p.*, r.name as role, r.id as role_id
+ FROM projects p
+ JOIN project_access pa ON p.id = pa.project_id
+ JOIN roles r ON pa.role_id = r.id
+ WHERE pa.user_id = ?
+ ''',
+ (user_id,)
+ )
+ return [dict(row) for row in cursor.fetchall()]
except Exception as e:
- logger.error(f"Database connection error: {str(e)}")
+ logger.error(f"Failed to get user projects: {str(e)}")
raise
-
+
+ # Script methods
def get_script(self, script_id: str) -> Optional[Dict[str, Any]]:
- """Get script details by ID with caching."""
+ """
+ Get a script by ID.
+ Uses caching to improve performance for frequently accessed scripts.
+ """
# Try to get from cache first
cached_script = self.script_cache.get(script_id)
if cached_script:
return cached_script
- # If not in cache, get from database
try:
with self._get_connection() as conn:
cursor = conn.execute(
- '''
- SELECT id, project_id, filename, content, created_at, updated_at, last_edited_by
- FROM scripts WHERE id = ?
- ''',
+ "SELECT * FROM scripts WHERE id = ?",
(script_id,)
)
- script = cursor.fetchone()
- if not script:
- return None
-
- script_dict = dict(script)
- script_dict["content_lines"] = script_dict["content"].splitlines()
-
- # Add to cache
- self.script_cache.set(script_id, script_dict)
- return script_dict
+ row = cursor.fetchone()
+ if row:
+ script_data = dict(row)
+ # Convert content to lines for easier processing if needed
+ # But store raw content in cache
+ if "content" in script_data:
+ script_data["content_lines"] = script_data["content"].splitlines()
+
+ # Cache the result
+ self.script_cache.set(script_id, script_data)
+ return script_data
+ return None
except Exception as e:
logger.error(f"Failed to get script: {str(e)}")
raise
-
- def get_script_by_filename(self, project_id: str, filename: str) -> Optional[Dict[str, Any]]:
- """Get script details by project ID and filename."""
+
+ def update_script(self, script_id: str, content: str, user_id: str) -> None:
+ """
+ Update script content and create a new version.
+ Invalidates the cache for this script.
+ """
try:
with self._get_connection() as conn:
- cursor = conn.execute(
- 'SELECT id, project_id, filename, content, created_at, updated_at, last_edited_by FROM scripts WHERE project_id = ? AND filename = ?',
- (project_id, filename)
+ # Update script content
+ conn.execute(
+ '''
+ UPDATE scripts
+ SET content = ?, last_edited_by = ?, updated_at = CURRENT_TIMESTAMP
+ WHERE id = ?
+ ''',
+ (content, user_id, script_id)
)
- script = cursor.fetchone()
- if not script:
- return None
- script_dict = dict(script)
- script_dict["content_lines"] = script_dict["content"].splitlines()
-
- return script_dict
- except Exception as e:
- logger.error(f"Failed to get script by filename: {str(e)}")
- raise
+ # Create new version
+ # Check if message column exists in versions table (it was missing in schema.sql but referenced in code)
+ # For compatibility with schema.sql which doesn't have 'message', we omit it if not needed
+ # Or check schema.sql content. It doesn't have message.
+ # So we should remove message from insert query.
+ conn.execute(
+ '''
+ INSERT INTO versions (id, script_id, content, created_by)
+ VALUES (?, ?, ?, ?)
+ ''',
+ (str(uuid.uuid4()), script_id, content, user_id)
+ )
+
+ # Invalidate cache
+ self.script_cache.delete(script_id)
- def delete_script(self, script_id: str) -> bool:
- """Delete a script and its versions."""
- # Invalidate cache
- self.script_cache.delete(script_id)
- try:
- with self._get_connection() as conn:
- conn.execute('BEGIN TRANSACTION')
- try:
- # Delete versions first due to foreign key constraint
- conn.execute('DELETE FROM versions WHERE script_id = ?', (script_id,))
- result = conn.execute('DELETE FROM scripts WHERE id = ?', (script_id,))
- deleted = result.rowcount > 0
- conn.execute('COMMIT')
-
- # Remove from cache if exists
- self.script_cache.delete(script_id)
-
- return deleted
- except Exception as e:
- conn.execute('ROLLBACK')
- raise
except Exception as e:
- logger.error(f"Failed to delete script: {str(e)}")
+ logger.error(f"Failed to update script: {str(e)}")
raise
- # TODO: Add methods for script searching and filtering - #issue/123
- def search_scripts(self, project_id: Optional[str] = None, query: Optional[str] = None, limit: int = 20) -> List[Dict]:
- """Search scripts by project and/or content keywords."""
+ def search_scripts(self, project_id: Optional[str] = None, query: Optional[str] = None, limit: int = 20) -> List[Dict[str, Any]]:
+ """Search for scripts."""
try:
+ sql = "SELECT id, project_id, filename, updated_at FROM scripts"
+ params = []
+ conditions = []
+
+ if project_id:
+ conditions.append("project_id = ?")
+ params.append(project_id)
+
+ if query:
+ conditions.append("(filename LIKE ? OR content LIKE ?)")
+ params.append(f"%{query}%")
+ params.append(f"%{query}%")
+
+ if conditions:
+ sql += " WHERE " + " AND ".join(conditions)
+
+ sql += " ORDER BY updated_at DESC LIMIT ?"
+ params.append(limit)
+
with self._get_connection() as conn:
- sql = """
- SELECT s.id, s.project_id, s.filename, s.updated_at,
- u.username as last_editor
- FROM scripts s
- LEFT JOIN users u ON s.last_edited_by = u.id
- WHERE 1=1
- """
- params = []
-
- if project_id:
- sql += " AND s.project_id = ?"
- params.append(project_id)
-
- if query:
- sql += " AND (s.filename LIKE ? OR s.content LIKE ?)"
- search = f"%{query}%"
- params.extend([search, search])
-
- sql += " ORDER BY s.updated_at DESC LIMIT ?"
- params.append(limit)
-
cursor = conn.execute(sql, params)
return [dict(row) for row in cursor.fetchall()]
except Exception as e:
logger.error(f"Failed to search scripts: {str(e)}")
raise
- # TODO: Add user management methods
- def create_user(self, username: str, email: str, password_hash: str) -> str:
- """Create a new user and return their ID."""
- user_id = str(uuid.uuid4())
- try:
- with sqlite3.connect(self.db_path) as conn:
- conn.execute(
- 'INSERT INTO users (id, username, email, password_hash) VALUES (?, ?, ?, ?)',
- (user_id, username, email, password_hash)
- )
- logger.info(f"Created user {username} with ID {user_id}")
- return user_id
- except sqlite3.IntegrityError:
- logger.error(f"Username or email already exists: {username}, {email}")
- raise ValueError("Username or email already exists")
- except Exception as e:
- logger.error(f"Failed to create user: {str(e)}")
- raise
-
- def get_user_by_username(self, username: str) -> Optional[Dict[str, Any]]:
- """Fetches a user by their username."""
- conn = self._get_connection()
- try:
- cursor = conn.cursor()
- cursor.execute("SELECT id, username, email, password_hash, created_at FROM users WHERE username = ?", (username,))
- user_data = cursor.fetchone()
- if user_data:
- return dict(user_data) # sqlite3.Row can be directly converted to dict
- return None
- except sqlite3.Error as e:
- logger.error(f"Database error when fetching user by username '{username}': {e}", exc_info=True)
- return None
- finally:
- if conn:
- conn.close()
+ # Collaboration methods
+ def acquire_node_lock(self, script_id: str, node_id: str, user_id: str, session_id: str, duration_seconds: int = 300) -> bool:
+ """
+ Acquire a lock on a node for editing.
+
+ Args:
+ script_id: The ID of the script containing the node
+ node_id: The ID of the node to lock
+ user_id: The ID of the user requesting the lock
+ session_id: The editing session ID
+ duration_seconds: How long the lock should be valid (default 5 minutes)
+
+ Returns:
+ True if lock acquired, False if already locked by someone else
+ """
+ expires_at = datetime.now() + timedelta(seconds=duration_seconds)
- def get_user_by_id(self, user_id: str) -> Optional[Dict[str, Any]]:
- """Get user details by ID."""
try:
with self._get_connection() as conn:
+ conn.execute("BEGIN TRANSACTION")
+
+ # Check for existing valid lock
cursor = conn.execute(
- 'SELECT id, username, email, password_hash, created_at FROM users WHERE id = ?',
- (user_id,)
+ '''
+ SELECT user_id, expires_at FROM node_locks
+ WHERE session_id = ? AND node_id = ?
+ ''',
+ (session_id, node_id)
)
- user = cursor.fetchone()
- if not user:
- return None
+ existing_lock = cursor.fetchone()
+
+ if existing_lock:
+ lock_user = existing_lock[0]
+ lock_expiry = datetime.fromisoformat(existing_lock[1]) if isinstance(existing_lock[1], str) else existing_lock[1]
+
+ # If locked by another user and not expired
+ if lock_user != user_id and lock_expiry > datetime.now():
+ conn.rollback()
+ return False
+
+ # If locked by same user or expired, update it
+ conn.execute(
+ '''
+ UPDATE node_locks
+ SET user_id = ?, expires_at = ?, locked_at = CURRENT_TIMESTAMP
+ WHERE session_id = ? AND node_id = ?
+ ''',
+ (user_id, expires_at, session_id, node_id)
+ )
+ else:
+ # Create new lock
+ conn.execute(
+ '''
+ INSERT INTO node_locks (id, session_id, user_id, node_id, expires_at)
+ VALUES (?, ?, ?, ?, ?)
+ ''',
+ (str(uuid.uuid4()), session_id, user_id, node_id, expires_at)
+ )
- return dict(user)
+ conn.commit()
+ return True
except Exception as e:
- logger.error(f"Failed to get user by ID: {str(e)}")
- raise
-
- def get_user_projects(self, user_id: str) -> List[Dict[str, Any]]:
- """Get all projects accessible by a user."""
+ logger.error(f"Failed to acquire lock: {str(e)}")
+ return False
+
+ def release_node_lock(self, script_id: str, node_id: str, user_id: str) -> bool:
+ """Release a lock on a node."""
try:
with self._get_connection() as conn:
- # Get projects owned by user
- owned_cursor = conn.execute(
- '''
- SELECT p.id, p.name, p.description, p.created_at, p.updated_at, p.owner_id,
- 'Owner' as role
- FROM projects p
- WHERE p.owner_id = ?
- ''',
- (user_id,)
- )
- owned_projects = [dict(row) for row in owned_cursor.fetchall()]
+ # We need to find the lock first to verify ownership
+ # Note: Schema uses session_id, but here we query by script_id logic?
+ # The node_locks table has session_id, not script_id directly.
+ # Assuming caller provides enough info or we look up session.
+ # Actually, node_locks is (node_id, session_id) unique.
+ # But we might have multiple sessions for a script?
+ # Typically one session per script per implementation?
+ # Let's assume we delete by node_id and user_id across any session for now,
+ # or we need session_id passed in.
+ # The previous implementation passed script_id which implies looking up sessions?
+ # Let's just try to delete where node_id and user_id matches.
- # Get projects shared with user
- shared_cursor = conn.execute(
+ conn.execute(
'''
- SELECT p.id, p.name, p.description, p.created_at, p.updated_at, p.owner_id,
- r.name as role
- FROM projects p
- JOIN project_access pa ON p.id = pa.project_id
- JOIN roles r ON pa.role_id = r.id
- WHERE pa.user_id = ?
+ DELETE FROM node_locks
+ WHERE node_id = ? AND user_id = ?
''',
- (user_id,)
+ (node_id, user_id)
)
- shared_projects = [dict(row) for row in shared_cursor.fetchall()]
-
- # Combine and ensure uniqueness (a user could be owner and also have explicit access)
- all_projects_dict = {p["id"]: p for p in owned_projects}
- for p in shared_projects:
- if p["id"] not in all_projects_dict:
- all_projects_dict[p["id"]] = p
- # Potentially update role if a more specific one is granted than just 'Owner'
- # For now, owner role takes precedence if listed as owned.
- # Or, if a user is an owner, their role is 'Owner' regardless of project_access entries.
-
- return list(all_projects_dict.values())
+ return True
except Exception as e:
- logger.error(f"Failed to get user projects: {str(e)}")
- raise
-
- def get_role_by_id(self, role_id: str) -> Optional[Dict[str, Any]]:
- """Get role details by role ID."""
+ logger.error(f"Failed to release lock: {str(e)}")
+ return False
+
+ def check_node_lock(self, script_id: str, node_id: str) -> Optional[str]:
+ """
+ Check if a node is locked and return the user_id if it is.
+ Returns None if not locked or lock expired.
+ """
try:
with self._get_connection() as conn:
+ # Need to join with sessions to filter by script_id if needed,
+ # or just check all locks for this node_id
cursor = conn.execute(
- "SELECT id, name, description FROM roles WHERE id = ?",
- (role_id,)
+ '''
+ SELECT user_id, expires_at FROM node_locks
+ WHERE node_id = ?
+ ''',
+ (node_id,)
)
- role = cursor.fetchone()
- return dict(role) if role else None
+ rows = cursor.fetchall()
+
+ current_time = datetime.now()
+ for row in rows:
+ expiry = row[1]
+ if isinstance(expiry, str):
+ expiry = datetime.fromisoformat(expiry)
+
+ if expiry > current_time:
+ return row[0]
+
+ return None
except Exception as e:
- logger.error(f"Failed to get role by ID '{role_id}': {str(e)}")
- raise
+ logger.error(f"Failed to check lock: {str(e)}")
+ return None
- def get_role_by_name(self, role_name: str) -> Optional[Dict[str, Any]]:
- """Get role details by role name."""
+ def refresh_node_lock(self, script_id: str, node_id: str, user_id: str, duration_seconds: int = 300) -> bool:
+ """Refresh an existing lock."""
+ expires_at = datetime.now() + timedelta(seconds=duration_seconds)
try:
with self._get_connection() as conn:
cursor = conn.execute(
- "SELECT id, name, description FROM roles WHERE name = ?",
- (role_name,)
+ '''
+ UPDATE node_locks
+ SET expires_at = ?
+ WHERE node_id = ? AND user_id = ?
+ ''',
+ (expires_at, node_id, user_id)
)
- role = cursor.fetchone()
- return dict(role) if role else None
+ return cursor.rowcount > 0
except Exception as e:
- logger.error(f"Failed to get role by name '{role_name}': {str(e)}")
- raise
-
+ logger.error(f"Failed to refresh lock: {str(e)}")
+ return False
+
def grant_project_access(self, project_id: str, user_id: str, role_id: str) -> bool:
- """Grant a user a specific role on a project."""
+ """Grant a user access to a project with a specific role."""
conn = self._get_connection()
try:
cursor = conn.cursor()
@@ -797,3 +769,58 @@ def track_script_edit(self, script_id: str, user_id: str, action_type: str, meta
# For now, we'll just log it
logger.info(f"Usage tracking: User {user_id} performed {action_type} on script {script_id}")
# TODO: Implement actual database tracking when analytics schema is added
+
+ def create_script(self, project_id: str, filename: str, content: str, user_id: str) -> str:
+ """
+ Create a new script in the database.
+
+ Args:
+ project_id: The ID of the project.
+ filename: The name of the script file.
+ content: The initial content of the script.
+ user_id: The ID of the user creating the script.
+
+ Returns:
+ The ID of the newly created script.
+ """
+ script_id = str(uuid.uuid4())
+ try:
+ with self._get_connection() as conn:
+ conn.execute(
+ '''
+ INSERT INTO scripts (id, project_id, filename, content, last_edited_by)
+ VALUES (?, ?, ?, ?, ?)
+ ''',
+ (script_id, project_id, filename, content, user_id)
+ )
+
+ # Create initial version
+ # Removed 'message' column to match schema.sql
+ conn.execute(
+ '''
+ INSERT INTO versions (id, script_id, content, created_by)
+ VALUES (?, ?, ?, ?)
+ ''',
+ (str(uuid.uuid4()), script_id, content, user_id)
+ )
+
+ return script_id
+ except Exception as e:
+ logger.error(f"Failed to create script: {str(e)}")
+ raise
+
+ def get_user_by_id(self, user_id: str) -> Optional[Dict[str, Any]]:
+ """Get user by ID."""
+ try:
+ with self._get_connection() as conn:
+ cursor = conn.execute(
+ "SELECT * FROM users WHERE id = ?",
+ (user_id,)
+ )
+ row = cursor.fetchone()
+ if row:
+ return dict(row)
+ return None
+ except Exception as e:
+ logger.error(f"Failed to get user by id: {str(e)}")
+ raise
diff --git a/backend/app/services/parser/__init__.py b/backend/app/services/parser/__init__.py
deleted file mode 100644
index e69de29..0000000
diff --git a/backend/app/services/parser/renpy_parser.py b/backend/app/services/parser/renpy_parser.py
deleted file mode 100644
index b6cb14b..0000000
--- a/backend/app/services/parser/renpy_parser.py
+++ /dev/null
@@ -1,598 +0,0 @@
-from enum import Enum
-from typing import List, Optional, Tuple
-import aiofiles
-
-
-class ChoiceNodeType(Enum):
- ACTION = "Action"
- LABEL_BLOCK = "LabelBlock"
- IF_BLOCK = "IfBlock"
- MENU_BLOCK = "MenuBlock"
- MENU_OPTION = "MenuOption"
-
-
-class ChoiceNode:
- """
- Represents a node in the choice tree structure of a RenPy script.
-
- Attributes:
- label_name (str): The name of the label.
- start_line (int): The starting line number in the script.
- end_line (int): The ending line number in the script.
- node_type (ChoiceNodeType): The type of the node.
- children (List[ChoiceNode]): Child nodes.
- false_branch (List[ChoiceNode]): The false branch children for if/elif/else statements.
- """
- def __init__(self,
- label_name: str = "",
- start_line: int = 0,
- end_line: int = 0,
- node_type: ChoiceNodeType = ChoiceNodeType.ACTION):
- self.label_name = label_name
- self.start_line = start_line
- self.end_line = end_line
- self.node_type = node_type
- self.children = []
- self.false_branch = []
-
-
-def _is_label(line: str) -> Tuple[bool, Optional[str]]:
- """
- Determines if a line is a label and extracts the label name.
-
- Args:
- line: The line to check.
-
- Returns:
- A tuple of (is_label, label_name)
- """
- line = line.strip()
- if line.startswith("label ") and line.endswith(':'):
- label_name = line[6:-1].strip()
- return True, label_name
- return False, None
-
-
-def _is_dialog_line(line: str) -> bool:
- """
- Checks if a line matches the RenPy dialog pattern:
- - Optional character name/expression followed by space
- - Text in quotes
- - Optional space at the end
- """
- line = line.strip()
-
- # Check if the line has quotes
- if '"' not in line:
- return False
-
- # Check if the line ends with a quote (ignoring trailing spaces)
- if not line.rstrip().endswith('"'):
- return False
-
- # Split at the first quote
- parts = line.split('"', 1)
-
- # If it starts with a quote, it's a dialog line without character name
- if line.startswith('"'):
- return True
-
- # There should be a space between character name and the opening quote
- character_part = parts[0].strip()
- return character_part.endswith(' ')
-
-
-def _remove_bracketed_content(text: str) -> str:
- """
- Removes all content enclosed in brackets, including the brackets themselves.
- For example, "1{brackets}23" becomes "123".
-
- Args:
- text: The input text to process
-
- Returns:
- Text with all bracketed content removed
- """
- result = ""
- bracket_level = 0
-
- for char in text:
- if char == '{':
- bracket_level += 1
- elif char == '}':
- bracket_level = max(0, bracket_level - 1) # Ensure we don't go negative
- elif bracket_level == 0:
- result += char
-
- return result
-
-
-class RenPyParser:
- """
- Asynchronous parser for RenPy code using recursive descent to build a choice tree.
- """
-
- def __init__(self):
- self.lines = []
-
- async def parse_async(self, script_path: str) -> ChoiceNode:
- """
- Parses the RenPy script asynchronously and returns the root choice node.
-
- Args:
- script_path: The path to the RenPy script file.
-
- Returns:
- The root choice node of the parsed script.
- """
- try:
- async with aiofiles.open(script_path, 'r', encoding='utf-8') as file:
- content = await file.read()
- self.lines = content.splitlines()
- except (FileNotFoundError, IOError) as e:
- raise IOError(f"Error reading script file: {e}")
-
- return self._parse_labels()
-
- def parse_text(self, content: str) -> ChoiceNode:
- """Parse RenPy script content provided as a string.
-
- This helper avoids having to write the content to a temporary file
- when we already have the full script in memory (for example after a
- node update). It mirrors :meth:`parse_async` but operates on the
- supplied text directly.
-
- Args:
- content: Full script content to parse.
-
- Returns:
- The root choice node of the parsed script.
- """
- self.lines = content.splitlines()
- return self._parse_labels()
-
- def _parse_labels(self) -> ChoiceNode:
- """
- Parses the labels in the RenPy script and builds a choice tree.
-
- Returns:
- root_node: The root choice node of the parsed script.
- """
- root_node = ChoiceNode(label_name="root", start_line=0)
- index = 0
-
- while index < len(self.lines):
- line = self.lines[index]
-
- label_info = _is_label(line)
- if label_info[0]:
- label_node = ChoiceNode(
- label_name=label_info[1],
- start_line=index,
- node_type=ChoiceNodeType.LABEL_BLOCK
- )
-
- index += 1
- label_child_node = ChoiceNode(start_line=index, node_type=ChoiceNodeType.ACTION)
-
- while True:
- result, index = self._parse_block(index, 1, label_child_node)
- if not result:
- break
- label_node.children.append(label_child_node)
- index += 1
- label_child_node = ChoiceNode(start_line=index, node_type=ChoiceNodeType.ACTION)
-
- # add check before label_child_node
- if label_child_node.end_line >= label_child_node.start_line:
- label_node.children.append(label_child_node)
-
- label_node.end_line = index - 1
- root_node.children.append(label_node)
- else:
- # skip lines before the next label
- index += 1
-
- return root_node
-
- def _parse_block(self, index: int, indent_level: int, current_node: ChoiceNode) -> Tuple[bool, int]:
- """
- Parses a block of lines with a given indentation level and updates the current node.
-
- Args:
- index: The current index in the lines array.
- indent_level: The expected indentation level.
- current_node: The current choice node being parsed.
-
- Returns:
- True if a new statement is encountered, otherwise False.
- """
- while index < len(self.lines):
- current_line = self.lines[index]
- current_indent = self._get_indent_level(current_line)
-
- if not current_line.strip():
- index += 1
- continue
-
- if current_indent < indent_level:
- index -= 1
- current_node.end_line = index
- return False, index
-
- if not self._is_a_statement(current_line.strip()):
- index += 1
- continue
-
- if current_node.start_line != index:
- index -= 1
- current_node.end_line = index
- return True, index
-
- trimmed_line = current_line.strip()
-
- if self._is_if_statement(trimmed_line):
- index = self._parse_statement(index, current_node, current_indent, ChoiceNodeType.IF_BLOCK)
- return True, index
-
- if self._is_menu_statement(trimmed_line):
- # Use the returned index to skip re-parsing menu lines
- index = self._parse_menu_block(index, current_node, current_indent)
- return True, index
-
- # Other statements can be handled here
- index += 1
-
- index -= 1
- current_node.end_line = index
- return False, index
-
- def _parse_statement(self, index: int, current_node: ChoiceNode,
- current_indent: int, node_type: ChoiceNodeType) -> int:
- """
- Parses a statement block and updates the current node.
-
- Args:
- index: The current index in the lines array.
- current_node: The current choice node being parsed.
- current_indent: The current indentation level.
- node_type: The type of the node being parsed.
-
- Returns:
- The updated index.
- """
- current_node.node_type = node_type
- current_node.end_line = index
- index += 1
- statement_node = ChoiceNode(start_line=index, node_type=ChoiceNodeType.ACTION)
- # Parse the 'true' branch
- while True:
- temp, index = self._parse_block(index, current_indent + 1, statement_node)
-
- # Only append nodes with valid content
- if statement_node.start_line <= statement_node.end_line:
- current_node.children.append(statement_node)
-
- if not temp:
- break
-
- index += 1
- statement_node = ChoiceNode(start_line=index, node_type=ChoiceNodeType.ACTION)
-
- # Check for 'elif' or 'else' at the same indentation level
- while index + 1 < len(self.lines):
- index += 1
- next_line = self.lines[index]
- next_indent = self._get_indent_level(next_line)
- next_line_trimmed = next_line.strip()
-
- if not next_line.strip():
- continue
-
- if next_line_trimmed.startswith('#'):
- continue
-
- if next_indent != current_indent:
- index -= 1
- break
-
- if self._is_elif_statement(next_line_trimmed):
- # Parse 'elif' as FalseBranch
- false_branch_node = ChoiceNode(start_line=index)
- index = self._parse_statement(index, false_branch_node, current_indent, ChoiceNodeType.IF_BLOCK)
- # Append to false_branch list instead of direct assignment
- current_node.false_branch.append(false_branch_node)
- return index
-
- # Handle 'else' statement
- if self._is_else_statement(next_line_trimmed):
- # Process all nodes in else branch
- index += 1
- while True:
- false_branch_node = ChoiceNode(start_line=index, node_type=ChoiceNodeType.ACTION)
- result, index = self._parse_block(index, current_indent + 1, false_branch_node)
-
- if false_branch_node.end_line >= false_branch_node.start_line:
- current_node.false_branch.append(false_branch_node)
-
- if not result:
- break
-
- index += 1
-
- return index
-
- index -= 1
- break
-
- return index
-
- def _parse_menu_block(self, index: int, menu_node: ChoiceNode, indent_level: int) -> int:
- """
- Parses a menu block and updates the menu node.
- Returns the updated index to avoid re-parsing the same lines.
- """
- menu_node.start_line = index
- menu_node.end_line = index
- menu_node.node_type = ChoiceNodeType.MENU_BLOCK
- index += 1
-
- while index < len(self.lines):
- line = self.lines[index]
- current_indent = self._get_indent_level(line)
-
- if not line.strip():
- index += 1
- continue
-
- if current_indent <= indent_level:
- index -= 1
- return index
-
- line = line.strip()
- if line.startswith('"') and line.endswith(':'):
- choice_node = ChoiceNode(
- label_name=line.rstrip(':').strip(),
- start_line=index,
- node_type=ChoiceNodeType.MENU_OPTION
- )
- index = self._parse_statement(index, choice_node, current_indent, ChoiceNodeType.MENU_OPTION)
- menu_node.children.append(choice_node)
- else:
- index += 1
-
- return index
-
- @staticmethod
- def _is_if_statement(line: str) -> bool:
- """
- Determines if a line is an if statement.
-
- Args:
- line: The line to check.
-
- Returns:
- True if the line is an if statement, otherwise False.
- """
- return line.lstrip().startswith("if ") and line.endswith(":")
-
- @staticmethod
- def _is_else_statement(line: str) -> bool:
- """
- Determines if a line is an else statement.
-
- Args:
- line: The line to check.
-
- Returns:
- True if the line is an else statement, otherwise False.
- """
- return line.lstrip().startswith("else") and line.endswith(':')
-
- @staticmethod
- def _is_elif_statement(line: str) -> bool:
- """
- Determines if a line is an elif statement.
-
- Args:
- line: The line to check.
-
- Returns:
- True if the line is an elif statement, otherwise False.
- """
- return line.lstrip().startswith("elif ") and line.endswith(':')
-
- @staticmethod
- def _is_a_statement(line: str) -> bool:
- """
- Determines if a line is a statement.
-
- Args:
- line: The line to check.
-
- Returns:
- True if the line is a statement, otherwise False.
- """
- trimmed_line = line.lstrip()
- return (trimmed_line.startswith("if ") or
- trimmed_line.startswith("elif ") or
- trimmed_line.startswith("menu"))
-
- @staticmethod
- def _is_menu_statement(line: str) -> bool:
- """
- Determines if a line is a menu statement.
-
- Args:
- line: The line to check.
-
- Returns:
- True if the line is a menu statement, otherwise False.
- """
- trimmed_line = line.lstrip()
- return trimmed_line.startswith("menu") and trimmed_line.endswith(":")
-
- @staticmethod
- def _get_indent_level(line: str) -> int:
- """
- Gets the indentation level of a line.
-
- Args:
- line: The line to check.
-
- Returns:
- The indentation level.
- """
- indent = 0
- tab_score = 0
-
- for char in line:
- if char == '\t':
- tab_score = 0
- indent += 1
- elif char == ' ':
- tab_score += 1
- if tab_score == 4:
- indent += 1
- tab_score = 0
- else:
- break
-
- return indent
-
- def _get_label_name(self, node: ChoiceNode) -> str:
- """
- Gets a descriptive label name for a node based on its content.
-
- For nodes with more than 4 lines, attempts to find dialog lines in RenPy format
- (character name followed by quoted text).
-
- Args:
- node: The node to get a label name for.
-
- Returns:
- A descriptive label name.
- """
- # Check if start_line or end_line is out of range
- if node.start_line >= len(self.lines) or node.end_line >= len(self.lines) or node.start_line < 0 or node.end_line < 0:
- return ""
-
- # If the first line ends with ":", it's a statement declaration
- if (node.start_line < len(self.lines) and
- self._is_a_statement(self.lines[node.start_line]) and
- self.lines[node.start_line].endswith(':')):
- return self.lines[node.start_line][:-1].strip()
-
- label_parts = []
- total_lines = node.end_line - node.start_line + 1
-
- # For small blocks (4 or fewer lines), include all non-empty lines
- if total_lines <= 4:
- for i in range(node.start_line, min(node.end_line + 1, len(self.lines))):
- if not self.lines[i].strip():
- continue
- label_parts.append(self.lines[i].strip())
- else:
- # Try to find dialog lines
- first_dialog_lines = []
- last_dialog_lines = []
-
- # Find first two dialog lines
- for i in range(node.start_line, min(node.end_line + 1, len(self.lines))):
- line = self.lines[i].strip()
- if not line:
- continue
-
- if _is_dialog_line(line):
- first_dialog_lines.append(line)
- if len(first_dialog_lines) >= 2:
- break
-
- # Find last two dialog lines (in correct order)
- for i in range(node.end_line, node.start_line - 1, -1):
- if i >= len(self.lines):
- continue
-
- line = self.lines[i].strip()
- if not line:
- continue
-
- if _is_dialog_line(line):
- last_dialog_lines.insert(0, line) # Insert at beginning to maintain order
- if len(last_dialog_lines) >= 2:
- break
-
- # Use dialog lines if we found any
- if first_dialog_lines or last_dialog_lines:
- label_parts.extend(first_dialog_lines)
-
- # Only add separator if we have both first and last sections
- if first_dialog_lines and last_dialog_lines and first_dialog_lines[-1] != last_dialog_lines[0]:
- label_parts.append("<...>")
-
- # Avoid duplicating lines
- for line in last_dialog_lines:
- if not first_dialog_lines or line not in first_dialog_lines:
- label_parts.append(line)
- else:
- # Fall back to using first/last 3 lines
- appended_lines = 0
- for i in range(node.start_line, min(node.end_line + 1, len(self.lines))):
- if not self.lines[i].strip():
- continue
- label_parts.append(self.lines[i].strip())
- appended_lines += 1
- if appended_lines >= 3:
- break
-
- label_parts.append("<...>")
-
- last_lines = []
- for i in range(node.end_line, node.start_line - 1, -1):
- if i >= len(self.lines) or not self.lines[i].strip():
- continue
- last_lines.insert(0, self.lines[i].strip())
- if len(last_lines) >= 3:
- break
-
- label_parts.extend(last_lines)
-
- # If we still have no label parts, just get all lines in the range to be safe
- if not label_parts:
- for i in range(node.start_line, min(node.end_line + 1, len(self.lines))):
- line = self.lines[i].strip()
- if line:
- label_parts.append(line)
-
- # Handle special commands like return/jump
- if not label_parts and node.start_line < len(self.lines):
- # Last attempt - use the single line directly
- line = self.lines[node.start_line].strip()
- if line:
- label_parts.append(line)
-
- label_text = "\n".join(label_parts)
-
- # Remove content within brackets (including the brackets)
- if not node.node_type == ChoiceNodeType.IF_BLOCK and not node.node_type == ChoiceNodeType.MENU_OPTION:
- label_text = _remove_bracketed_content(label_text)
-
- # If the label is still empty or very short, try every line in the range
- if len(label_text) < 20:
- combined_text = []
-
- # Important: Include ALL lines in the node's range
- for i in range(node.start_line, min(node.end_line + 1, len(self.lines))):
- if i < len(self.lines): # Double-check index
- line = self.lines[i].strip()
- if line:
- combined_text.append(line)
-
- if combined_text:
- label_text = "\n".join(combined_text)
-
- # Truncate to 100 characters if text is too long
- if len(label_text) > 100:
- label_text = label_text[:97] + "..."
-
- return label_text
\ No newline at end of file
diff --git a/backend/app/services/parser/test_renpy_parser.py b/backend/app/services/parser/test_renpy_parser.py
deleted file mode 100644
index 45de7cd..0000000
--- a/backend/app/services/parser/test_renpy_parser.py
+++ /dev/null
@@ -1,197 +0,0 @@
-import os
-import pytest
-import tempfile
-import textwrap
-from pathlib import Path
-from .renpy_parser import RenPyParser, ChoiceNodeType
-
-@pytest.fixture
-def sample_renpy_script():
- """Create a temporary RenPy script file for testing."""
- script_content = """
- label start:
- "Это начало истории."
-
- menu:
- "Выбрать первый путь":
- jump path_one
- "Выбрать второй путь":
- jump path_two
-
- label path_one:
- "Вы выбрали первый путь."
-
- if condition:
- "Что-то происходит при выполнении условия."
- else:
- "Что-то происходит, если условие не выполнено."
-
- return
-
- label path_two:
- "Вы выбрали второй путь."
- return
- """
-
- with tempfile.NamedTemporaryFile(suffix='.rpy', delete=False, mode='w', encoding='utf-8') as f:
- f.write(textwrap.dedent(script_content))
-
- yield f.name
-
- os.unlink(f.name)
-
-@pytest.mark.asyncio
-async def test_parse_async_basic(sample_renpy_script):
- """Test basic parsing functionality."""
- parser = RenPyParser()
- root_node = await parser.parse_async(sample_renpy_script)
-
- assert root_node is not None
- assert root_node.label_name == "root"
-
- labels = [child.label_name for child in root_node.children
- if child.node_type == ChoiceNodeType.LABEL_BLOCK]
-
- assert len(root_node.children) > 1 # Should have start
- assert "start" in labels
- assert "path_one" in labels
- assert "path_two" in labels
-
-@pytest.mark.asyncio
-async def test_parse_async_nonexistent_file():
- """Test error handling when file is not found."""
- parser = RenPyParser()
- with pytest.raises(IOError):
- await parser.parse_async("nonexistent_file.rpy")
-
-@pytest.mark.asyncio
-async def test_parse_menu_structure(sample_renpy_script):
- """Test if menu structures are parsed correctly."""
- parser = RenPyParser()
- root_node = await parser.parse_async(sample_renpy_script)
-
- start_label = next((child for child in root_node.children
- if child.label_name == "start"), None)
- assert start_label is not None
-
- menu_node = None
- for action_node in start_label.children:
- if action_node.node_type == ChoiceNodeType.MENU_BLOCK:
- menu_node = action_node
- break
-
- assert menu_node is not None
- assert menu_node.node_type == ChoiceNodeType.MENU_BLOCK
- assert len(menu_node.children) == 2
- assert all(child.label_name for child in menu_node.children), "Меню содержит опцию без имени"
- menu_options = [child.label_name.strip('"') for child in menu_node.children]
- assert "Выбрать первый путь" in menu_options
- assert "Выбрать второй путь" in menu_options
-
-@pytest.mark.asyncio
-async def test_parse_if_else_structure(sample_renpy_script):
- """Test if conditional structures are parsed correctly."""
- parser = RenPyParser()
- root_node = await parser.parse_async(sample_renpy_script)
-
- path_one_label = next((child for child in root_node.children
- if child.label_name == "path_one"), None)
- assert path_one_label is not None
-
- if_node = None
- for action_node in path_one_label.children:
- if action_node.node_type == ChoiceNodeType.IF_BLOCK:
- if_node = action_node
- break
-
- assert if_node is not None
- # Check if else branch exists - false_branch is now a list
- assert if_node.false_branch is not None
- assert len(if_node.false_branch) > 0
-
- # Get the first node in the false branch
- false_branch_node = if_node.false_branch[0]
- assert false_branch_node.node_type == ChoiceNodeType.ACTION
-
-
-def test_comment_before_else_does_not_break_false_branch():
- parser = RenPyParser()
- script_content = textwrap.dedent(
- """
- label comment_if:
- if condition:
- "True branch"
- # some note about the branch
- else:
- "False branch"
- """
- )
-
- root_node = parser.parse_text(script_content)
-
- label_node = next(
- (child for child in root_node.children if child.label_name == "comment_if"),
- None
- )
- assert label_node is not None
-
- if_node = next(
- (child for child in label_node.children if child.node_type == ChoiceNodeType.IF_BLOCK),
- None
- )
- assert if_node is not None
-
- assert isinstance(if_node.false_branch, list)
- assert len(if_node.false_branch) == 1
- false_branch_node = if_node.false_branch[0]
- assert false_branch_node.node_type == ChoiceNodeType.ACTION
-
-@pytest.mark.asyncio
-async def test_label_with_only_actions_no_extra_node():
- """
- Tests that a label with only actions has one child action node
- spanning the entire label body, and that no extra nodes are created.
- This specifically tests for a bug where an extra node was created
- or the action node's line numbers were corrupted.
- """
- script_content = """
-label simple_label:
- "Action 1"
- "Action 2"
- jump next_one
-
-label next_one:
- return
- """
- with tempfile.NamedTemporaryFile(suffix='.rpy', delete=False, mode='w', encoding='utf-8') as f:
- f.write(textwrap.dedent(script_content))
- filepath = f.name
-
- try:
- parser = RenPyParser()
- root_node = await parser.parse_async(filepath)
-
- simple_label = next((c for c in root_node.children if c.label_name == "simple_label"), None)
- assert simple_label is not None
- assert simple_label.node_type == ChoiceNodeType.LABEL_BLOCK
-
- # The label should contain exactly one child node of type ACTION
- assert len(simple_label.children) == 1
- action_node = simple_label.children[0]
- assert action_node.node_type == ChoiceNodeType.ACTION
-
- # Script lines from dedent:
- # 1: label simple_label:
- # 2: "Action 1"
- # 3: "Action 2"
- # 4: jump next_one
- # 5:
- # 6: label next_one:
-
- # The action node should start after the label definition
- assert action_node.start_line == 2
-
- # And end on the line before the next label (including empty lines)
- assert action_node.end_line == 5
- finally:
- os.unlink(filepath)
diff --git a/backend/tests/test_project_management.py b/backend/tests/test_project_management.py
index d40922c..f481037 100644
--- a/backend/tests/test_project_management.py
+++ b/backend/tests/test_project_management.py
@@ -1,42 +1,79 @@
-import unittest
import pytest
-import os
import tempfile
-import sqlite3
+import os
import uuid
import json
-from pathlib import Path
-import shutil
+from unittest.mock import MagicMock, AsyncMock, patch
from fastapi.testclient import TestClient
-from fastapi import Depends
-from unittest.mock import patch
-import builtins
-
from app.main import app
from app.services.database import DatabaseService
from app.api.routes.auth import get_current_user, oauth2_scheme
+from fastapi import Depends
+
+client = TestClient(app)
+# --- Mocks and Helpers ---
test_db_for_auth = None
-app_db_service = None
+@pytest.fixture(scope="module")
+def temp_db():
+ global test_db_for_auth
+ with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as temp_db_file:
+ temp_db_path = temp_db_file.name
+
+ os.environ['DATABASE_PATH'] = temp_db_path
+ db_service = DatabaseService()
+ test_db_for_auth = db_service
+
+ yield db_service
+
+ # Cleanup
+ if os.path.exists(temp_db_path):
+ os.remove(temp_db_path)
+ test_db_for_auth = None
+
+@pytest.fixture
+def auth_token(temp_db):
+ """Creates a user and returns a valid auth token."""
+ from app.services.auth import AuthService
+
+ user_id = str(uuid.uuid4())
+ username = "admin"
+ # Check if user exists first to avoid unique constraint error in repeated tests using same db fixture
+ exists = temp_db.get_user_by_username(username)
+ if not exists:
+ with temp_db._get_connection() as conn:
+ conn.execute(
+ "INSERT INTO users (id, username, email, password_hash) VALUES (?, ?, ?, ?)",
+ (user_id, username, "admin@example.com", "b2")
+ )
+ conn.commit()
+ else:
+ user_id = exists["id"]
+
+ auth_service = AuthService(temp_db)
+ token = auth_service.create_access_token({"sub": user_id, "username": username})
+ return token
+
+# Mock get_current_user dependency to use our test DB for user validation
async def override_get_current_user(token: str = Depends(oauth2_scheme)):
from app.services.auth import AuthService
from fastapi import HTTPException, status
-
+
global test_db_for_auth
if test_db_for_auth is None:
raise ValueError("Test database not initialized")
-
+
auth_service = AuthService(test_db_for_auth)
-
+
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
-
+
payload = auth_service.decode_token(token)
if payload is None:
raise credentials_exception
@@ -48,351 +85,100 @@ async def override_get_current_user(token: str = Depends(oauth2_scheme)):
user = test_db_for_auth.get_user_by_id(user_id)
if user is None:
raise credentials_exception
-
- return user
-
-
-from app.api.routes import projects
-from app.api.routes import scripts
+ return user
app.dependency_overrides[get_current_user] = override_get_current_user
-client = TestClient(app)
-
-
-@pytest.fixture(autouse=True)
-def override_app_db_service(temp_db):
- """Override the app's database service with our test database instance."""
-
- original_projects_db = projects.db_service
- original_scripts_db = scripts.db_service
-
-
- projects.db_service = temp_db
- scripts.db_service = temp_db
-
-
- def get_db_service():
- return temp_db
-
-
- if hasattr(projects, 'get_db_service'):
- original_get_db_service = projects.get_db_service
- projects.get_db_service = get_db_service
-
-
- original_get_project = projects.get_project
-
- async def debug_get_project(*args, **kwargs):
- print("DEBUG: Using test_db in get_project!")
- print(f"DEBUG: test_db object id: {id(temp_db)}")
- print(f"DEBUG: projects.db_service object id: {id(projects.db_service)}")
- try:
- return await original_get_project(*args, **kwargs)
- except Exception as e:
- print(f"DEBUG: get_project error: {e}")
- raise
-
- projects.get_project = debug_get_project
-
- yield
-
-
- projects.db_service = original_projects_db
- scripts.db_service = original_scripts_db
- projects.get_project = original_get_project
- if hasattr(projects, 'get_db_service'):
- projects.get_db_service = original_get_db_service
-
-@pytest.fixture
-def temp_db():
- """Create a temporary database for testing."""
- global test_db_for_auth
-
-
- original_db_path = os.environ.get('DATABASE_PATH', None)
-
-
- temp_dir = tempfile.TemporaryDirectory()
- db_path = os.path.join(temp_dir.name, 'test.db')
- os.environ['DATABASE_PATH'] = db_path
-
-
- backend_dir = Path(__file__).parent.parent
- schema_path = backend_dir / 'database' / 'schema.sql'
-
-
- os.makedirs(os.path.dirname(schema_path), exist_ok=True)
-
-
- if not schema_path.exists():
-
- project_root = backend_dir.parent
- source_schema = project_root / 'database' / 'schema.sql'
- if source_schema.exists():
- print(f"Copying schema from {source_schema} to {schema_path}")
- os.makedirs(schema_path.parent, exist_ok=True)
- with open(source_schema, 'r') as src, open(schema_path, 'w') as dest:
- dest.write(src.read())
-
-
- db_service = None
- try:
- db_service = DatabaseService()
-
- test_db_for_auth = db_service
-
-
- with db_service._get_connection() as conn:
-
- cursor = conn.execute("SELECT COUNT(*) FROM roles")
- if cursor.fetchone()[0] == 0:
-
- roles = [
- ('role_owner', 'Owner', 'Full control over the project'),
- ('role_editor', 'Editor', 'Can edit project content'),
- ('role_viewer', 'Viewer', 'Can view project content'),
- ]
- conn.executemany(
- "INSERT INTO roles (id, name, description) VALUES (?, ?, ?)",
- roles
- )
- conn.commit()
-
- yield db_service
- finally:
-
- if db_service:
- db_service.close()
-
-
- test_db_for_auth = None
-
-
- if original_db_path:
- os.environ['DATABASE_PATH'] = original_db_path
- else:
- os.environ.pop('DATABASE_PATH', None)
-
-
- try:
- temp_dir.cleanup()
- except (PermissionError, OSError) as e:
- print(f"Warning: Could not clean temporary directory: {e}")
-
-
-@pytest.fixture
-def auth_token(temp_db):
- """Create a test user and return auth token."""
-
- username = "admin"
- password = "adminadmin"
- password_hash = "$2b$12$.R2kUy.ihZcA6D.YTvLdEu7h9PAs66LtNlMtcxhzM/9T.xOyTfUPO"
- email = "admin@example.com"
-
-
- with temp_db._get_connection() as conn:
-
- cursor = conn.execute("SELECT id FROM users WHERE username = ?", (username,))
- existing_user = cursor.fetchone()
-
- if existing_user:
- user_id = existing_user['id']
- else:
-
- user_id = str(uuid.uuid4())
- conn.execute(
- "INSERT INTO users (id, username, email, password_hash) VALUES (?, ?, ?, ?)",
- (user_id, username, email, password_hash)
- )
- conn.commit()
-
-
- from app.services.auth import AuthService
- from datetime import timedelta
-
-
- auth_service = AuthService(temp_db)
-
-
- token_data = {
- "sub": user_id,
- "username": username
- }
-
-
- token = auth_service.create_access_token(
- data=token_data,
- expires_delta=timedelta(hours=24)
- )
-
- return token
class TestProjectManagement:
- """Test project management functionality."""
def test_create_project(self, auth_token):
- """Test creating a project through API."""
+ """Test creating a new project."""
headers = {"Authorization": f"Bearer {auth_token}"}
-
-
- project_name = "Test Project"
- project_description = "A test project"
response = client.post(
"/api/projects/",
headers=headers,
- json={"name": project_name, "description": project_description}
+ json={"name": "New Project", "description": "A test project"}
)
-
- assert response.status_code == 200, f"Failed to create project: {response.text}"
- project_data = response.json()
- project_id = project_data["id"]
- assert project_data["name"] == project_name
-
-
- response = client.get(f"/api/projects/{project_id}", headers=headers)
- assert response.status_code == 200, f"Failed to get project: {response.text}"
+ assert response.status_code == 200
data = response.json()
- assert data["id"] == project_id, "Project ID doesn't match"
- assert data["name"] == project_name, "Project name doesn't match"
-
+ assert data["name"] == "New Project"
+ assert "id" in data
+ assert "owner_id" in data
+
def test_list_projects(self, auth_token):
- """Test listing user's projects."""
+ """Test listing projects for a user."""
headers = {"Authorization": f"Bearer {auth_token}"}
+ # Create a few projects
+ client.post("/api/projects/", headers=headers, json={"name": "Project 1"})
+ client.post("/api/projects/", headers=headers, json={"name": "Project 2"})
- for i in range(3):
- response = client.post(
- "/api/projects/",
- headers=headers,
- json={"name": f"Test Project {i}", "description": f"Description {i}"}
- )
- assert response.status_code == 200, f"Failed to create project: {response.text}"
-
-
- response = client.get(
- "/api/projects/",
- headers=headers
- )
- assert response.status_code == 200, f"Failed to list projects: {response.text}"
- projects = response.json()
- assert len(projects) >= 3, "Not all created projects returned"
+ response = client.get("/api/projects/", headers=headers)
+ assert response.status_code == 200
+ data = response.json()
+ assert len(data) >= 2
+ assert any(p["name"] == "Project 1" for p in data)
def test_get_project(self, auth_token):
- """Test getting a single project through API."""
+ """Test retrieving a specific project."""
headers = {"Authorization": f"Bearer {auth_token}"}
+ # Create project
+ create_resp = client.post("/api/projects/", headers=headers, json={"name": "Target Project"})
+ project_id = create_resp.json()["id"]
- project_name = "Project Details"
- project_description = "Test getting project details"
-
- response = client.post(
- "/api/projects/",
- headers=headers,
- json={"name": project_name, "description": project_description}
- )
- assert response.status_code == 200, f"Failed to create project: {response.text}"
- project_data = response.json()
- project_id = project_data["id"]
-
-
- response = client.get(
- f"/api/projects/{project_id}",
- headers=headers
- )
-
-
- print(f"API Response status: {response.status_code}")
- print(f"API Response body: {response.text}")
-
- assert response.status_code == 200, f"Failed to get project: {response.text}"
- project = response.json()
- assert project["id"] == project_id, "Project ID doesn't match"
- assert project["name"] == project_name, "Project name doesn't match"
- assert project["description"] == project_description, "Project description doesn't match"
+ # Get project
+ response = client.get(f"/api/projects/{project_id}", headers=headers)
+ assert response.status_code == 200
+ data = response.json()
+ assert data["id"] == project_id
+ assert data["name"] == "Target Project"
- @patch('app.api.routes.scripts.parse_script')
- def test_script_upload_with_project(self, mock_parse_script, auth_token):
- """Test uploading a script with explicit project selection."""
-
- mock_parse_script.return_value = {
- "script_id": str(uuid.uuid4()),
- "filename": "test_script.rpy",
- "tree": {"id": "mock_tree", "node_type": "LabelBlock", "label_name": "root", "children": []}
- }
-
+ def test_script_upload_integration(self, auth_token):
+ """Test uploading a script (integration style)."""
headers = {"Authorization": f"Bearer {auth_token}"}
+ # 1. Create project
+ create_resp = client.post("/api/projects/", headers=headers, json={"name": "Upload Project"})
+ assert create_resp.status_code == 200
+ project_id = create_resp.json()["id"]
- project_name = "Script Project"
- project_description = "Project for script testing"
-
- response = client.post(
- "/api/projects/",
- headers=headers,
- json={"name": project_name, "description": project_description}
- )
- assert response.status_code == 200, f"Failed to create project: {response.text}"
- project_data = response.json()
- project_id = project_data["id"]
-
+ # 2. Upload script
+ script_content = "label start:\n return"
- script_content = """
-label start:
- "Hello, this is a test script."
- return
-"""
with tempfile.NamedTemporaryFile(suffix=".rpy", delete=False, mode="w+") as temp_file:
temp_file.write(script_content)
temp_file_path = temp_file.name
-
- try:
+ try:
with open(temp_file_path, "rb") as f:
response = client.post(
"/api/scripts/parse",
headers=headers,
files={"file": f},
- data={"project_id": project_id}
+ data={"project_id": project_id}
)
- assert response.status_code == 200, f"Script upload failed: {response.text}"
+ assert response.status_code == 200
data = response.json()
- assert "script_id" in data, "Script ID not returned"
- assert "filename" in data, "Filename not returned"
- assert "tree" in data, "Parsed tree not returned"
-
+ assert "script_id" in data
+ assert data["content"] == script_content
+ # tree should NOT be in data anymore
+ assert "tree" not in data
- response = client.get(
- f"/api/projects/{project_id}",
- headers=headers
- )
- project = response.json()
- assert "scripts" in project, "Scripts not included in project details"
- assert len(project["scripts"]) > 0, "No scripts associated with project"
- script_found = False
- for script in project["scripts"]:
- if script["id"] == data["script_id"]:
- script_found = True
- break
- assert script_found, "Uploaded script not found in project scripts"
-
finally:
-
- try:
- os.unlink(temp_file_path)
- except:
- pass
+ if os.path.exists(temp_file_path):
+ os.remove(temp_file_path)
def test_upload_without_project_id_fails(self, auth_token):
- """Test that uploading a script without a project_id fails."""
+ """Test that uploading without project_id fails (validation error)."""
headers = {"Authorization": f"Bearer {auth_token}"}
- script_content = "label start:\n \"This should fail.\"\n return"
with tempfile.NamedTemporaryFile(suffix=".rpy", delete=False, mode="w+") as temp_file:
- temp_file.write(script_content)
+ temp_file.write("label start:\n return")
temp_file_path = temp_file.name
try:
@@ -400,245 +186,98 @@ def test_upload_without_project_id_fails(self, auth_token):
response = client.post(
"/api/scripts/parse",
headers=headers,
- files={"file": f},
- data={}
+ files={"file": f}
+ # Missing data={"project_id": ...}
)
-
+ # 422 Unprocessable Entity due to missing form field
assert response.status_code == 422
- assert response.json()["detail"][0]['loc'] == ["body", "project_id"]
- assert response.json()["detail"][0]['msg'] == "Field required"
-
finally:
- try:
- os.unlink(temp_file_path)
- except:
- pass
+ if os.path.exists(temp_file_path):
+ os.remove(temp_file_path)
def test_share_project(self, auth_token, temp_db):
- """Test sharing a project with another user via API."""
+ """Test sharing a project with another user."""
+ headers_user1 = {"Authorization": f"Bearer {auth_token}"}
- headers_user1 = {"Authorization": f"Bearer {auth_token}"} # Token for the project owner
+ # 1. Create project
+ create_resp = client.post("/api/projects/", headers=headers_user1, json={"name": "Shared Project"})
+ project_id = create_resp.json()["id"]
- # 1. Create a project as user1
- project_name = "ProjectToShare"
- create_project_resp = client.post(
- "/api/projects/",
- headers=headers_user1,
- json={"name": project_name, "description": "A project to test sharing"}
- )
- assert create_project_resp.status_code == 200, f"Failed to create project: {create_project_resp.text}"
- project_id = create_project_resp.json()["id"]
-
- # 2. Create a second user (user2) directly in the database for testing
- from app.services.auth import AuthService # Moved import here
+ # 2. Create user2
user2_id = str(uuid.uuid4())
user2_username = "test_share_user2"
- with temp_db._get_connection() as conn:
- conn.execute(
- "INSERT INTO users (id, username, email, password_hash) VALUES (?, ?, ?, ?)",
- (user2_id, user2_username, "share_user2@example.com", "$2b$12$dummyhashforshare")
- )
- conn.commit()
+ # Check uniqueness
+ if not temp_db.get_user_by_username(user2_username):
+ with temp_db._get_connection() as conn:
+ conn.execute(
+ "INSERT INTO users (id, username, email, password_hash) VALUES (?, ?, ?, ?)",
+ (user2_id, user2_username, "share_user2@example.com", "hash")
+ )
+ conn.commit()
+ else:
+ user2_id = temp_db.get_user_by_username(user2_username)["id"]
- # 3. User1 shares the project with User2 via API, assigning 'role_viewer'
- # Ensure 'role_viewer' is a valid role ID from your temp_db setup
- role_to_assign = "role_viewer"
- share_payload = {"user_id": user2_username, "role": role_to_assign}
-
+ # 3. Share project
share_resp = client.post(
f"/api/projects/{project_id}/share",
- headers=headers_user1, # User1 (owner) performs the share action
- json=share_payload
+ headers=headers_user1,
+ json={"user_id": user2_username, "role": "role_viewer"}
)
- assert share_resp.status_code == 200, f"API call to share project failed: {share_resp.text}. Payload: {share_payload}"
+ assert share_resp.status_code == 200
- # 4. Verify User2 can now access the project
- auth_service_user2 = AuthService(temp_db)
- token_user2 = auth_service_user2.create_access_token({"sub": user2_id, "username": user2_username})
+ # 4. Verify access
+ from app.services.auth import AuthService
+ auth_service = AuthService(temp_db)
+ token_user2 = auth_service.create_access_token({"sub": user2_id, "username": user2_username})
headers_user2 = {"Authorization": f"Bearer {token_user2}"}
- get_project_resp_user2 = client.get(
- f"/api/projects/{project_id}",
- headers=headers_user2 # User2 attempts to access
- )
- assert get_project_resp_user2.status_code == 200, f"User2 failed to access shared project: {get_project_resp_user2.text}"
- project_details_for_user2 = get_project_resp_user2.json()
- assert project_details_for_user2["id"] == project_id
- # Optionally, assert role if project details include it for the current user
- # assert project_details_for_user2.get("role") == role_to_assign # Or however role is exposed
-
- # 5. Verify the project sharing record in the database
- with temp_db._get_connection() as conn:
- cursor = conn.execute(
- "SELECT role_id FROM project_access WHERE project_id = ? AND user_id = ?",
- (project_id, user2_id)
- )
- project_user_entry = cursor.fetchone()
- assert project_user_entry is not None, "Project sharing record not found in database after API call"
- assert project_user_entry["role_id"] == role_to_assign, f"Incorrect role_id in database. Expected {role_to_assign}, got {project_user_entry['role_id']}"
+ get_resp = client.get(f"/api/projects/{project_id}", headers=headers_user2)
+ assert get_resp.status_code == 200
def test_share_project_with_role_name(self, auth_token, temp_db):
- """Test sharing a project with another user via API using the role NAME."""
+ """Test sharing a project using role name (e.g. 'Viewer')."""
+ headers_user1 = {"Authorization": f"Bearer {auth_token}"}
- headers_user1 = {"Authorization": f"Bearer {auth_token}"} # Token for the project owner
+ create_resp = client.post("/api/projects/", headers=headers_user1, json={"name": "RoleName Share"})
+ project_id = create_resp.json()["id"]
- # 1. Create a project as user1
- project_name = "ProjectToShareByName"
- create_project_resp = client.post(
- "/api/projects/",
- headers=headers_user1,
- json={"name": project_name, "description": "A project to test sharing by role name"}
- )
- assert create_project_resp.status_code == 200, f"Failed to create project: {create_project_resp.text}"
- project_id = create_project_resp.json()["id"]
-
- # 2. Create a second user (user2)
- from app.services.auth import AuthService
user2_id = str(uuid.uuid4())
- user2_username = "test_share_user2_by_name"
- with temp_db._get_connection() as conn:
- conn.execute(
- "INSERT INTO users (id, username, email, password_hash) VALUES (?, ?, ?, ?)",
- (user2_id, user2_username, "share_user2_name@example.com", "$2b$12$dummyhashforname")
- )
- conn.commit()
-
- # 3. User1 shares the project with User2 via API, assigning role by NAME
- role_name_to_send_in_payload = "Viewer" # This is the role NAME
- expected_role_id_in_db = "role_viewer" # This is the corresponding ID
+ user2_username = "role_name_user"
+ if not temp_db.get_user_by_username(user2_username):
+ with temp_db._get_connection() as conn:
+ conn.execute(
+ "INSERT INTO users (id, username, email, password_hash) VALUES (?, ?, ?, ?)",
+ (user2_id, user2_username, "role_name@example.com", "hash")
+ )
+ conn.commit()
+ else:
+ user2_id = temp_db.get_user_by_username(user2_username)["id"]
- share_payload = {"user_id": user2_username, "role": role_name_to_send_in_payload}
-
share_resp = client.post(
f"/api/projects/{project_id}/share",
headers=headers_user1,
- json=share_payload
+ json={"user_id": user2_username, "role": "Viewer"}
)
+ assert share_resp.status_code == 200
- # IDEAL BEHAVIOR: Backend should accept name, find ID, and return 200
- # or return 4xx if names are not supported / name is invalid.
- # A 500 here indicates a bug in handling this case.
- assert share_resp.status_code == 200, f"API call to share project with role name failed: {share_resp.text}. Payload: {share_payload}"
-
- # 4. Verify User2 can access
- auth_service_user2 = AuthService(temp_db)
- token_user2 = auth_service_user2.create_access_token({"sub": user2_id, "username": "user_for_delete_test"})
- headers_user2 = {"Authorization": f"Bearer {token_user2}"}
-
- get_project_resp_user2 = client.get(
- f"/api/projects/{project_id}",
- headers=headers_user2
- )
- assert get_project_resp_user2.status_code == 200, f"User2 failed to access project shared by role name: {get_project_resp_user2.text}"
- project_details_for_user2 = get_project_resp_user2.json()
- assert project_details_for_user2["id"] == project_id
-
- # 5. Verify the correct role_id was stored in the database
+ # Verify DB
with temp_db._get_connection() as conn:
- cursor = conn.execute(
+ row = conn.execute(
"SELECT role_id FROM project_access WHERE project_id = ? AND user_id = ?",
(project_id, user2_id)
- )
- project_user_entry = cursor.fetchone()
- assert project_user_entry is not None, "Project sharing record not found in database after API call (role name)"
- assert project_user_entry["role_id"] == expected_role_id_in_db, f"Incorrect role_id in database. Expected {expected_role_id_in_db}, got {project_user_entry['role_id']}"
-
-
- def test_create_script_endpoint(self, auth_token):
- """Test creating a script via the projects/{project_id}/scripts endpoint."""
-
- headers = {"Authorization": f"Bearer {auth_token}"}
- resp = client.post("/api/projects/", headers=headers, json={"name":"ScriptProj"})
- project_id = resp.json()["id"]
-
-
- payload = {"filename":"a.rpy", "content":"label a:\n return"}
- create_resp = client.post(
- f"/api/projects/{project_id}/scripts",
- headers=headers,
- json=payload
- )
- assert create_resp.status_code == 200
- script_id = create_resp.json()["id"]
-
-
- proj = client.get(f"/api/projects/{project_id}", headers=headers).json()
- assert any(s["id"] == script_id for s in proj["scripts"])
+ ).fetchone()
+ assert row["role_id"] == "role_viewer"
def test_delete_project_endpoint(self, auth_token, temp_db):
- """Test deleting a project via the projects/{project_id} endpoint."""
+ """Test deleting a project."""
headers = {"Authorization": f"Bearer {auth_token}"}
-
- # 1. Create a project
- create_resp = client.post("/api/projects/", headers=headers, json={"name": "ProjectToDelete"})
- assert create_resp.status_code == 200
- project_to_delete_id = create_resp.json()["id"]
- owner_id = create_resp.json()["owner_id"] # Assuming owner_id is returned
-
- # 2. Delete the project as the owner
- delete_resp = client.delete(f"/api/projects/{project_to_delete_id}", headers=headers)
- assert delete_resp.status_code == 200
- assert delete_resp.json()["message"] == f"Project {project_to_delete_id} deleted successfully."
-
- # 3. Verify the project is no longer accessible
- get_resp = client.get(f"/api/projects/{project_to_delete_id}", headers=headers)
- assert get_resp.status_code == 404 # Or appropriate error for not found/access denied after deletion
-
- # 4. Attempt to delete a non-existent project
- non_existent_project_id = str(uuid.uuid4())
- delete_non_existent_resp = client.delete(f"/api/projects/{non_existent_project_id}", headers=headers)
- assert delete_non_existent_resp.status_code == 404
-
- # 5. Test deletion permission: Create another project
- create_resp_2 = client.post("/api/projects/", headers=headers, json={"name": "ProjectToTestPermissions"})
- assert create_resp_2.status_code == 200
- project_id_perms_test = create_resp_2.json()["id"]
-
- # 6. Create a second user and token
- from app.services.auth import AuthService
- second_user_id = str(uuid.uuid4())
- with temp_db._get_connection() as conn:
- conn.execute(
- "INSERT INTO users (id, username, email, password_hash) VALUES (?, ?, ?, ?)",
- (second_user_id, "user_for_delete_test", "user_del_test@example.com", "$2b$12$dummyhash")
- )
- conn.commit()
-
- auth_service_2 = AuthService(temp_db)
- token_2 = auth_service_2.create_access_token({"sub": second_user_id, "username": "user_for_delete_test"})
- headers_2 = {"Authorization": f"Bearer {token_2}"}
-
- # 7. Attempt to delete the project with the second user's token (should fail)
- delete_permission_resp = client.delete(f"/api/projects/{project_id_perms_test}", headers=headers_2)
- assert delete_permission_resp.status_code == 403 # Forbidden
-
- # 8. Verify project still exists (was not deleted by non-owner)
- get_resp_after_failed_delete = client.get(f"/api/projects/{project_id_perms_test}", headers=headers) # Check with owner token
- assert get_resp_after_failed_delete.status_code == 200
-
-
- def _create_mock_tree(self):
- """Create a mock parse tree for testing."""
- from app.services.parser.renpy_parser import ChoiceNode, ChoiceNodeType
-
-
- root = ChoiceNode("root", ChoiceNodeType.LABEL_BLOCK)
- root.start_line = 0
- root.end_line = 5
- start_node = ChoiceNode("start", ChoiceNodeType.LABEL_BLOCK)
- start_node.start_line = 1
- start_node.end_line = 4
-
- dialogue = ChoiceNode('"Hello, this is a test script."', ChoiceNodeType.ACTION)
- dialogue.start_line = 2
- dialogue.end_line = 2
-
- return_node = ChoiceNode("return", ChoiceNodeType.ACTION)
- return_node.start_line = 3
- return_node.end_line = 3
+ create_resp = client.post("/api/projects/", headers=headers, json={"name": "Del Project"})
+ assert create_resp.status_code == 200
+ project_id = create_resp.json()["id"]
- start_node.children = [dialogue, return_node]
- root.children = [start_node]
+ del_resp = client.delete(f"/api/projects/{project_id}", headers=headers)
+ assert del_resp.status_code == 200
- return root
+ get_resp = client.get(f"/api/projects/{project_id}", headers=headers)
+ assert get_resp.status_code == 404
diff --git a/frontend/src/services/api.ts b/frontend/src/services/api.ts
index b0ce7af..6c3b0d8 100644
--- a/frontend/src/services/api.ts
+++ b/frontend/src/services/api.ts
@@ -1,185 +1,103 @@
-///
-import axios, { AxiosError } from 'axios';
-
-export interface ParsedScriptResponse {
- script_id: string;
- filename: string;
- tree: any;
-}
-
-// Интерфейс для ответа при получении содержимого узла
-export interface NodeContentResponse {
- content: string;
- start_line: number;
- end_line: number;
-}
-
-// Интерфейс для ответа при обновлении содержимого узла
-export interface UpdateNodeResponse {
- message: string;
- line_diff: number;
- content: string;
-}
-
-export interface InsertNodeResponse {
- start_line: number;
- end_line: number;
- line_count: number;
- tree: any;
-}
-
-const runtimeConfig = typeof window !== 'undefined' ? (window as any).RUNTIME_CONFIG : undefined;
-const effectiveApiUrl = runtimeConfig?.VITE_API_URL || import.meta.env.VITE_API_URL;
-// Log the URL being used to help debug
-console.log('API Base URL configured:', effectiveApiUrl);
-
-export const apiClient = axios.create({
- baseURL: effectiveApiUrl, // Use the env variable directly
+import axios, { AxiosInstance, AxiosError } from 'axios';
+import {
+ ParsedScriptResponse,
+ NodeContentResponse,
+ UpdateNodeResponse,
+ InsertNodeResponse
+} from './api.d';
+import { RenPyParser } from '../utils/renpyParser';
+
+// Create a configured axios instance
+const apiClient: AxiosInstance = axios.create({
+ baseURL: import.meta.env.VITE_API_URL || 'http://localhost:8000/api',
headers: {
'Content-Type': 'application/json',
},
});
-// Add auth token to all requests if available
-apiClient.interceptors.request.use((config) => {
- const token = typeof localStorage !== 'undefined' ? localStorage.getItem('auth_token') : null;
- if (token && config.headers) {
- config.headers.Authorization = `Bearer ${token}`;
+// Add a request interceptor to include the auth token
+apiClient.interceptors.request.use(
+ (config) => {
+ const token = localStorage.getItem('token');
+ if (token) {
+ config.headers.Authorization = `Bearer ${token}`;
+ }
+ return config;
+ },
+ (error) => {
+ return Promise.reject(error);
}
- return config;
-}, (error) => {
- return Promise.reject(error);
-});
+);
+
+// --- API Functions ---
/**
- * Parses an uploaded RenPy script file.
- * @param file - The .rpy file to parse.
- * @param projectId - Optional project ID to associate the script with.
- * @returns The parsed script data (script_id, filename, tree).
+ * Parses a script file.
+ * Now handles client-side parsing if backend returns raw content.
*/
export const parseScript = async (file: File, projectId?: string): Promise => {
const formData = new FormData();
formData.append('file', file);
-
- // Add project_id if provided
if (projectId) {
formData.append('project_id', projectId);
}
- const targetUrl = apiClient.defaults.baseURL + '/scripts/parse'; // Construct full URL for logging
- console.log(`[API Request] POST ${targetUrl} with file: ${file.name}${projectId ? ` for project: ${projectId}` : ''}`);
-
try {
+ // If backend returns just content, we parse it here.
+ // If backend returns tree, we use it (backward compatibility).
const response = await apiClient.post('/scripts/parse', formData, {
headers: {
'Content-Type': 'multipart/form-data',
},
});
- console.log('[API Response] parseScript successful:', response.data);
+
+ if (response.data.content && !response.data.tree) {
+ console.log('[API] Client-side parsing uploaded script...');
+ const parser = new RenPyParser();
+ response.data.tree = parser.parse(response.data.content);
+ }
+
return response.data;
} catch (error) {
- // --- Enhanced Error Logging ---
- console.error('[API Error] Failed during parseScript call.');
- console.error('Target URL:', targetUrl);
- console.error('File Name:', file.name);
-
const axiosError = error as AxiosError;
-
- if (axiosError.response) {
- // The request was made and the server responded with a status code
- // that falls out of the range of 2xx
- console.error('Error Response Data:', axiosError.response.data);
- console.error('Error Response Status:', axiosError.response.status);
- console.error('Error Response Headers:', axiosError.response.headers);
- } else if (axiosError.request) {
- // The request was made but no response was received
- console.error('Error Request:', axiosError.request);
- console.error('No response received from server. Check network connection and backend status.');
- } else {
- // Something happened in setting up the request that triggered an Error
- console.error('Error Message:', axiosError.message);
- }
- console.error('Full Error Object:', error);
- // --- End Enhanced Error Logging ---
-
- // Re-throw a more informative error if possible, otherwise the original
- throw axiosError.response?.data || new Error(`Failed to parse script '${file.name}'. Status: ${axiosError.response?.status || 'unknown'}. ${axiosError.message}`);
+ console.error('Error uploading/parsing script:', axiosError);
+ throw axiosError.response?.data || new Error('Failed to parse script');
}
};
/**
* Creates a new script file with default content.
- * @param filename - The desired filename.
- * @param projectId - Optional project ID to associate the script with.
- * @returns The parsed script data for the new file.
*/
export const createNewScript = async (filename: string = 'new_script.rpy', projectId?: string): Promise => {
- console.log(`[API Action] Attempting to create new script: ${filename}${projectId ? ` for project: ${projectId}` : ''}`);
- const defaultContent = 'label Start:\n return'; // Basic Renpy script
+ const defaultContent = 'label Start:\n return';
const blob = new Blob([defaultContent], { type: 'text/plain' });
const file = new File([blob], filename, { type: 'text/plain' });
- // parseScript already has enhanced logging, so errors here will be detailed
- try {
- const result = await parseScript(file, projectId);
- console.log(`[API Action] Successfully created and parsed new script: ${filename}`);
- return result;
- } catch (error) {
- console.error(`[API Error] Failed during createNewScript for ${filename}. Error originated from parseScript call.`);
- // Re-throw the error caught from parseScript
- throw error; // Re-throw the detailed error from parseScript
- }
+ return parseScript(file, projectId);
};
/**
- * Получает содержимое узла на основе его начальной и конечной строки.
- * @param scriptId - ID скрипта, к которому принадлежит узел.
- * @param startLine - Начальная строка узла.
- * @param endLine - Конечная строка узла.
- * @returns Обещание с содержимым узла.
+ * Gets node content.
*/
export const getNodeContent = async (
scriptId: string,
startLine: number,
endLine: number
): Promise => {
- const targetUrl = `${apiClient.defaults.baseURL}/scripts/node-content/${scriptId}`;
- console.log(`[API Request] GET ${targetUrl} for node lines ${startLine}-${endLine}`);
-
try {
const response = await apiClient.get(`/scripts/node-content/${scriptId}`, {
params: { start_line: startLine, end_line: endLine },
});
- console.log('[API Response] getNodeContent successful:', response.data);
return response.data;
} catch (error) {
- console.error('[API Error] Failed during getNodeContent call.');
- console.error('Script ID:', scriptId);
- console.error('Start line:', startLine);
- console.error('End line:', endLine);
-
const axiosError = error as AxiosError;
-
- if (axiosError.response) {
- console.error('Error Response Data:', axiosError.response.data);
- console.error('Error Response Status:', axiosError.response.status);
- } else if (axiosError.request) {
- console.error('Error Request:', axiosError.request);
- } else {
- console.error('Error Message:', axiosError.message);
- }
-
- throw axiosError.response?.data || new Error(`Failed to get node content. Status: ${axiosError.response?.status || 'unknown'}. ${axiosError.message}`);
+ console.error('Error getting node content:', axiosError);
+ throw axiosError.response?.data || new Error('Failed to get node content');
}
};
/**
- * Обновляет содержимое узла в скрипте.
- * @param scriptId - ID скрипта, к которому принадлежит узел.
- * @param startLine - Начальная строка узла (до редактирования).
- * @param endLine - Конечная строка узла (до редактирования).
- * @param content - Новое содержимое узла.
- * @returns Обещание с информацией об обновлении.
+ * Updates node content.
*/
export const updateNodeContent = async (
scriptId: string,
@@ -187,36 +105,17 @@ export const updateNodeContent = async (
endLine: number,
content: string
): Promise => {
- const targetUrl = `${apiClient.defaults.baseURL}/scripts/update-node/${scriptId}`;
- console.log(`[API Request] POST ${targetUrl} to update node lines ${startLine}-${endLine}`);
-
try {
const response = await apiClient.post(
`/scripts/update-node/${scriptId}`,
- { content }, // Отправляем содержимое в теле запроса
- { params: { start_line: startLine, end_line: endLine } } // Строки в параметрах запроса
+ { content },
+ { params: { start_line: startLine, end_line: endLine } }
);
- console.log('[API Response] updateNodeContent successful:', response.data);
return response.data;
} catch (error) {
- console.error('[API Error] Failed during updateNodeContent call.');
- console.error('Script ID:', scriptId);
- console.error('Start line:', startLine);
- console.error('End line:', endLine);
- console.error('Content length:', content.length);
-
const axiosError = error as AxiosError;
-
- if (axiosError.response) {
- console.error('Error Response Data:', axiosError.response.data);
- console.error('Error Response Status:', axiosError.response.status);
- } else if (axiosError.request) {
- console.error('Error Request:', axiosError.request);
- } else {
- console.error('Error Message:', axiosError.message);
- }
-
- throw axiosError.response?.data || new Error(`Failed to update node content. Status: ${axiosError.response?.status || 'unknown'}. ${axiosError.message}`);
+ console.error('Error updating node content:', axiosError);
+ throw axiosError.response?.data || new Error('Failed to update node content');
}
};
@@ -226,101 +125,56 @@ export const insertNode = async (
nodeType: string,
content: string,
): Promise => {
- const targetUrl = `${apiClient.defaults.baseURL}/scripts/insert-node/${scriptId}`;
- console.log(`[API Request] POST ${targetUrl} to insert ${nodeType} at line ${insertionLine}`);
-
try {
const response = await apiClient.post(
`/scripts/insert-node/${scriptId}`,
{ content, node_type: nodeType },
{ params: { insertion_line: insertionLine } },
);
- console.log('[API Response] insertNode successful:', response.data);
+ // If insertNode returns updated content (not tree), we might need to handle it?
+ // Usually insertion triggers a reload or local update.
+ // For now, assume callers reload or backend returns tree (if not changed yet).
return response.data;
} catch (error) {
- console.error('[API Error] Failed during insertNode call.');
- console.error('Script ID:', scriptId);
- console.error('Insertion line:', insertionLine);
- console.error('Node type:', nodeType);
-
const axiosError = error as AxiosError;
-
- if (axiosError.response) {
- console.error('Error Response Data:', axiosError.response.data);
- console.error('Error Response Status:', axiosError.response.status);
- } else if (axiosError.request) {
- console.error('Error Request:', axiosError.request);
- } else {
- console.error('Error Message:', axiosError.message);
- }
-
- throw axiosError.response?.data || new Error(`Failed to insert node. Status: ${axiosError.response?.status || 'unknown'}. ${axiosError.message}`);
+ console.error('Error inserting node:', axiosError);
+ throw axiosError.response?.data || new Error('Failed to insert node');
}
};
/**
- * Получает полное содержимое скрипта для сохранения на локальный диск.
- * @param scriptId - ID скрипта.
- * @returns Обещание с полным содержимым скрипта.
+ * Gets full script content.
*/
export const getScriptContent = async (scriptId: string): Promise => {
- const targetUrl = `${apiClient.defaults.baseURL}/scripts/download/${scriptId}`;
- console.log(`[API Request] GET ${targetUrl} for full script content`);
-
try {
const response = await apiClient.get<{content: string, filename: string}>(`/scripts/download/${scriptId}`);
- console.log('[API Response] getScriptContent successful');
return response.data.content;
} catch (error) {
- console.error('[API Error] Failed during getScriptContent call.');
- console.error('Script ID:', scriptId);
-
const axiosError = error as AxiosError;
-
- if (axiosError.response) {
- console.error('Error Response Data:', axiosError.response.data);
- console.error('Error Response Status:', axiosError.response.status);
- } else if (axiosError.request) {
- console.error('Error Request:', axiosError.request);
- } else {
- console.error('Error Message:', axiosError.message);
- }
-
- throw axiosError.response?.data || new Error(`Failed to get full script content. Status: ${axiosError.response?.status || 'unknown'}. ${axiosError.message}`);
+ console.error('Error downloading script:', axiosError);
+ throw axiosError.response?.data || new Error('Failed to download script');
}
};
/**
- * Loads an existing script by its ID and returns parsed data.
- * @param scriptId - The ID of the script to load.
- * @returns The script content and parsed tree data.
+ * Loads an existing script.
*/
export const loadExistingScript = async (scriptId: string): Promise => {
- const targetUrl = `${apiClient.defaults.baseURL}/scripts/load/${scriptId}`;
- console.log(`[API Request] GET ${targetUrl} to load existing script`);
-
try {
const response = await apiClient.get(`/scripts/load/${scriptId}`);
- console.log('[API Response] loadExistingScript successful:', response.data);
- return response.data;
- } catch (error) {
- console.error('[API Error] Failed during loadExistingScript call.');
- console.error('Script ID:', scriptId);
- const axiosError = error as AxiosError;
-
- if (axiosError.response) {
- console.error('Error Response Data:', axiosError.response.data);
- console.error('Error Response Status:', axiosError.response.status);
- } else if (axiosError.request) {
- console.error('Error Request:', axiosError.request);
- } else {
- console.error('Error Message:', axiosError.message);
+ if (response.data.content && !response.data.tree) {
+ console.log('[API] Client-side parsing existing script...');
+ const parser = new RenPyParser();
+ response.data.tree = parser.parse(response.data.content);
}
- throw axiosError.response?.data || new Error(`Failed to load script. Status: ${axiosError.response?.status || 'unknown'}. ${axiosError.message}`);
+ return response.data;
+ } catch (error) {
+ const axiosError = error as AxiosError;
+ console.error('Error loading script:', axiosError);
+ throw axiosError.response?.data || new Error('Failed to load script');
}
};
-// Export the API client
export default apiClient;
diff --git a/frontend/src/utils/__tests__/branching.test.ts b/frontend/src/utils/__tests__/branching.test.ts
deleted file mode 100644
index 16e49d1..0000000
--- a/frontend/src/utils/__tests__/branching.test.ts
+++ /dev/null
@@ -1,54 +0,0 @@
-import { describe, expect, it } from 'vitest';
-import { buildBranchSnippet } from '../branching';
-
-describe('buildBranchSnippet', () => {
- it('adds an extra indent level for menu snippets', () => {
- const snippet = buildBranchSnippet({
- type: 'menu',
- indent: '',
- options: [
- { text: 'First option' },
- ],
- });
-
- const lines = snippet.split('\n');
- expect(lines[0]).toBe(' menu:');
- expect(lines[1]).toBe(' "First option":');
- expect(lines[2]).toBe(' # TODO: Add action for "First option"');
- expect(lines[3]).toBe(' pass');
- });
-
- it('adds an extra indent level for conditional snippets', () => {
- const snippet = buildBranchSnippet({
- type: 'if',
- indent: '',
- branches: [
- { kind: 'if', condition: 'player_has_item' },
- { kind: 'else' },
- ],
- });
-
- const lines = snippet.split('\n');
- expect(lines[0]).toBe(' if player_has_item:');
- expect(lines[1]).toBe(' # PSEUDO CONDITION: player_has_item');
- expect(lines[2]).toBe(' pass');
- expect(lines[4]).toBe(' else:');
- expect(lines[5]).toBe(' # TODO: Handle else branch');
- });
-
- it('nests menu snippets relative to existing indentation', () => {
- const snippet = buildBranchSnippet({
- type: 'menu',
- indent: ' ',
- options: [
- { text: 'Nested option' },
- ],
- });
-
- const lines = snippet.split('\n');
- expect(lines[0]).toBe(' menu:');
- expect(lines[1]).toBe(' "Nested option":');
- expect(lines[2]).toBe(' # TODO: Add action for "Nested option"');
- expect(lines[3]).toBe(' pass');
- });
-});
diff --git a/frontend/src/utils/renpyParser.ts b/frontend/src/utils/renpyParser.ts
new file mode 100644
index 0000000..67176e9
--- /dev/null
+++ b/frontend/src/utils/renpyParser.ts
@@ -0,0 +1,305 @@
+import { ParsedNodeData } from './parsedNodeTypes';
+
+export enum ChoiceNodeType {
+ ACTION = 'Action',
+ LABEL_BLOCK = 'LabelBlock',
+ IF_BLOCK = 'IfBlock',
+ MENU_BLOCK = 'MenuBlock',
+ MENU_OPTION = 'MenuOption',
+}
+
+// Helper to generate unique IDs
+const generateId = (): string => {
+ if (typeof crypto !== 'undefined' && crypto.randomUUID) {
+ return crypto.randomUUID();
+ }
+ return 'node-' + Math.random().toString(36).substr(2, 9) + Date.now().toString(36);
+};
+
+export class RenPyParser {
+ private lines: string[] = [];
+
+ public parse(text: string): ParsedNodeData {
+ this.lines = text.split(/\r?\n/);
+
+ // Root node
+ const root: ParsedNodeData = {
+ id: generateId(),
+ node_type: ChoiceNodeType.ACTION,
+ label_name: 'root',
+ start_line: 0,
+ end_line: 0,
+ children: [],
+ };
+
+ let index = 0;
+ while (index < this.lines.length) {
+ const line = this.lines[index];
+ const strippedLine = line.trim();
+
+ if (!strippedLine || strippedLine.startsWith('#')) {
+ index++;
+ continue;
+ }
+
+ // Check for Label definition
+ const [isLabel, labelName] = this._isLabel(line);
+ if (isLabel && labelName) {
+ const labelNode: ParsedNodeData = {
+ id: generateId(),
+ node_type: ChoiceNodeType.LABEL_BLOCK,
+ label_name: labelName,
+ start_line: index,
+ end_line: index,
+ children: [],
+ };
+
+ index = this._parseBlock(index + 1, labelNode, 1);
+ labelNode.end_line = index > labelNode.start_line! ? index - 1 : labelNode.start_line;
+
+ root.children?.push(labelNode);
+ } else {
+ index++;
+ }
+ }
+
+ return root;
+ }
+
+ private _parseBlock(index: number, parentNode: ParsedNodeData, indentLevel: number): number {
+ while (index < this.lines.length) {
+ let line = this.lines[index];
+ let currentIndent = this._getIndentLevel(line);
+
+ // Skip empty lines
+ if (!line.trim()) {
+ index++;
+ continue;
+ }
+
+ if (currentIndent < indentLevel) {
+ // Indent mismatch (dedent).
+ // Backtrack over any trailing empty lines we might have consumed
+ while (index > 0 && !this.lines[index-1].trim()) {
+ index--;
+ }
+ return index;
+ }
+
+ if (this._isIfStatement(line)) {
+ const ifNode: ParsedNodeData = {
+ id: generateId(),
+ node_type: ChoiceNodeType.IF_BLOCK,
+ label_name: '',
+ start_line: index,
+ end_line: index,
+ children: [],
+ false_branch: [],
+ };
+ index = this._parseIfBlock(index, ifNode, currentIndent);
+ parentNode.children?.push(ifNode);
+ }
+ else if (this._isMenuStatement(line)) {
+ const menuNode: ParsedNodeData = {
+ id: generateId(),
+ node_type: ChoiceNodeType.MENU_BLOCK,
+ label_name: '',
+ start_line: index,
+ end_line: index,
+ children: [],
+ };
+ index = this._parseMenuBlock(index, menuNode, currentIndent);
+ parentNode.children?.push(menuNode);
+ }
+ else {
+ // Generic Action / Dialog
+ const actionNode: ParsedNodeData = {
+ id: generateId(),
+ node_type: ChoiceNodeType.ACTION,
+ label_name: '',
+ start_line: index,
+ end_line: index,
+ children: [],
+ };
+
+ while (index < this.lines.length) {
+ line = this.lines[index];
+
+ if (!line.trim()) {
+ index++;
+ continue;
+ }
+
+ const nextIndent = this._getIndentLevel(line);
+ if (nextIndent !== indentLevel) {
+ // Indent mismatch. Do NOT backtrack empty lines here.
+ break;
+ }
+
+ if (this._isIfStatement(line) || this._isMenuStatement(line) || this._isLabel(line)[0]) {
+ break;
+ }
+
+ index++;
+ }
+
+ // Set end_line to the last consumed line (index - 1)
+ actionNode.end_line = index - 1;
+
+ if (actionNode.start_line <= actionNode.end_line) {
+ parentNode.children?.push(actionNode);
+ }
+ }
+ }
+ return index;
+ }
+
+ private _parseIfBlock(index: number, ifNode: ParsedNodeData, indentLevel: number): number {
+ ifNode.start_line = index;
+ index++;
+
+ // Parse true branch
+ index = this._parseBlock(index, ifNode, indentLevel + 1);
+
+ while (index < this.lines.length) {
+ let line = this.lines[index];
+ if (!line.trim()) {
+ index++;
+ continue;
+ }
+
+ const currentIndent = this._getIndentLevel(line);
+ if (currentIndent !== indentLevel) {
+ break;
+ }
+
+ if (this._isElifStatement(line)) {
+ const elifNode: ParsedNodeData = {
+ id: generateId(),
+ node_type: ChoiceNodeType.IF_BLOCK,
+ label_name: '',
+ start_line: index,
+ end_line: index,
+ children: [],
+ false_branch: [],
+ };
+
+ if (!ifNode.false_branch) ifNode.false_branch = [];
+ ifNode.false_branch.push(elifNode);
+
+ index++;
+ index = this._parseBlock(index, elifNode, indentLevel + 1);
+
+ ifNode = elifNode;
+
+ } else if (this._isElseStatement(line)) {
+ index++;
+
+ const dummyParent: ParsedNodeData = { children: [] };
+ index = this._parseBlock(index, dummyParent, indentLevel + 1);
+
+ if (!ifNode.false_branch) ifNode.false_branch = [];
+ if (dummyParent.children) {
+ ifNode.false_branch.push(...dummyParent.children);
+ }
+ break;
+ } else {
+ break;
+ }
+ }
+
+ return index;
+ }
+
+ private _parseMenuBlock(index: number, menuNode: ParsedNodeData, indentLevel: number): number {
+ menuNode.start_line = index;
+ index++;
+
+ while (index < this.lines.length) {
+ const line = this.lines[index];
+ const currentIndent = this._getIndentLevel(line);
+
+ if (!line.trim()) {
+ index++;
+ continue;
+ }
+
+ if (currentIndent <= indentLevel) {
+ while (index > 0 && !this.lines[index-1].trim()) {
+ index--;
+ }
+ return index;
+ }
+
+ const stripped = line.trim();
+ if (stripped.startsWith('"') && stripped.endsWith(':')) {
+ const labelName = stripped.replace(/:$/, '').trim();
+ const optionNode: ParsedNodeData = {
+ id: generateId(),
+ node_type: ChoiceNodeType.MENU_OPTION,
+ label_name: labelName,
+ start_line: index,
+ end_line: index,
+ children: [],
+ };
+
+ index++;
+ index = this._parseBlock(index, optionNode, currentIndent + 1);
+
+ menuNode.children?.push(optionNode);
+ } else {
+ index++;
+ }
+ }
+ return index;
+ }
+
+ // Helpers
+ private _isLabel(line: string): [boolean, string | null] {
+ const trimmed = line.trim();
+ if (trimmed.startsWith('label ') && trimmed.endsWith(':')) {
+ return [true, trimmed.substring(6, trimmed.length - 1).trim()];
+ }
+ return [false, null];
+ }
+
+ private _isIfStatement(line: string): boolean {
+ const trimmed = line.trim();
+ return trimmed.startsWith('if ') && trimmed.endsWith(':');
+ }
+
+ private _isElifStatement(line: string): boolean {
+ const trimmed = line.trim();
+ return trimmed.startsWith('elif ') && trimmed.endsWith(':');
+ }
+
+ private _isElseStatement(line: string): boolean {
+ const trimmed = line.trim();
+ return trimmed.startsWith('else') && trimmed.endsWith(':');
+ }
+
+ private _isMenuStatement(line: string): boolean {
+ const trimmed = line.trim();
+ return trimmed.startsWith('menu') && trimmed.endsWith(':');
+ }
+
+ private _getIndentLevel(line: string): number {
+ let indent = 0;
+ let tabScore = 0;
+ for (const char of line) {
+ if (char === '\t') {
+ tabScore = 0;
+ indent += 1;
+ } else if (char === ' ') {
+ tabScore += 1;
+ if (tabScore === 4) {
+ indent += 1;
+ tabScore = 0;
+ }
+ } else {
+ break;
+ }
+ }
+ return indent;
+ }
+}