3
3
from app .services .github_service import GitHubService
4
4
from app .services .claude_service import ClaudeService
5
5
from app .core .limiter import limiter
6
- import os
7
- from app .prompts import SYSTEM_FIRST_PROMPT , SYSTEM_SECOND_PROMPT , SYSTEM_THIRD_PROMPT , ADDITIONAL_SYSTEM_INSTRUCTIONS_PROMPT
6
+ from app .prompts import (
7
+ SYSTEM_FIRST_PROMPT ,
8
+ SYSTEM_SECOND_PROMPT ,
9
+ SYSTEM_THIRD_PROMPT ,
10
+ ADDITIONAL_SYSTEM_INSTRUCTIONS_PROMPT ,
11
+ )
8
12
from anthropic ._exceptions import RateLimitError
9
13
from pydantic import BaseModel
10
14
from functools import lru_cache
@@ -29,11 +33,7 @@ def get_cached_github_data(username: str, repo: str):
29
33
file_tree = github_service .get_github_file_paths_as_list (username , repo )
30
34
readme = github_service .get_github_readme (username , repo )
31
35
32
- return {
33
- "default_branch" : default_branch ,
34
- "file_tree" : file_tree ,
35
- "readme" : readme
36
- }
36
+ return {"default_branch" : default_branch , "file_tree" : file_tree , "readme" : readme }
37
37
38
38
39
39
class ApiRequest (BaseModel ):
@@ -51,7 +51,13 @@ async def generate(request: Request, body: ApiRequest):
51
51
if len (body .instructions ) > 1000 :
52
52
return {"error" : "Instructions exceed maximum length of 1000 characters" }
53
53
54
- if body .repo in ["fastapi" , "streamlit" , "flask" , "api-analytics" , "monkeytype" ]:
54
+ if body .repo in [
55
+ "fastapi" ,
56
+ "streamlit" ,
57
+ "flask" ,
58
+ "api-analytics" ,
59
+ "monkeytype" ,
60
+ ]:
55
61
return {"error" : "Example repos cannot be regenerated" }
56
62
57
63
# Get cached github data
@@ -71,7 +77,7 @@ async def generate(request: Request, body: ApiRequest):
71
77
return {
72
78
"error" : f"File tree and README combined exceeds token limit (50,000). Current size: { token_count } tokens. This GitHub repository is too large for my wallet, but you can continue by providing your own Anthropic API key." ,
73
79
"token_count" : token_count ,
74
- "requires_api_key" : True
80
+ "requires_api_key" : True ,
75
81
}
76
82
elif token_count > 200000 :
77
83
return {
@@ -82,20 +88,22 @@ async def generate(request: Request, body: ApiRequest):
82
88
first_system_prompt = SYSTEM_FIRST_PROMPT
83
89
third_system_prompt = SYSTEM_THIRD_PROMPT
84
90
if body .instructions :
85
- first_system_prompt = first_system_prompt + \
86
- "\n " + ADDITIONAL_SYSTEM_INSTRUCTIONS_PROMPT
87
- third_system_prompt = third_system_prompt + \
88
- "\n " + ADDITIONAL_SYSTEM_INSTRUCTIONS_PROMPT
91
+ first_system_prompt = (
92
+ first_system_prompt + "\n " + ADDITIONAL_SYSTEM_INSTRUCTIONS_PROMPT
93
+ )
94
+ third_system_prompt = (
95
+ third_system_prompt + "\n " + ADDITIONAL_SYSTEM_INSTRUCTIONS_PROMPT
96
+ )
89
97
90
98
# get the explanation for sysdesign from claude
91
99
explanation = claude_service .call_claude_api (
92
100
system_prompt = first_system_prompt ,
93
101
data = {
94
102
"file_tree" : file_tree ,
95
103
"readme" : readme ,
96
- "instructions" : body .instructions
104
+ "instructions" : body .instructions ,
97
105
},
98
- api_key = body .api_key
106
+ api_key = body .api_key ,
99
107
)
100
108
101
109
# Check for BAD_INSTRUCTIONS response
@@ -104,18 +112,14 @@ async def generate(request: Request, body: ApiRequest):
104
112
105
113
full_second_response = claude_service .call_claude_api (
106
114
system_prompt = SYSTEM_SECOND_PROMPT ,
107
- data = {
108
- "explanation" : explanation ,
109
- "file_tree" : file_tree
110
- }
115
+ data = {"explanation" : explanation , "file_tree" : file_tree },
111
116
)
112
117
113
118
# Extract component mapping from the response
114
119
start_tag = "<component_mapping>"
115
120
end_tag = "</component_mapping>"
116
121
component_mapping_text = full_second_response [
117
- full_second_response .find (start_tag ):
118
- full_second_response .find (end_tag )
122
+ full_second_response .find (start_tag ) : full_second_response .find (end_tag )
119
123
]
120
124
121
125
# get mermaid.js code from claude
@@ -124,8 +128,8 @@ async def generate(request: Request, body: ApiRequest):
124
128
data = {
125
129
"explanation" : explanation ,
126
130
"component_mapping" : component_mapping_text ,
127
- "instructions" : body .instructions
128
- }
131
+ "instructions" : body .instructions ,
132
+ },
129
133
)
130
134
131
135
# Check for BAD_INSTRUCTIONS response
@@ -134,18 +138,14 @@ async def generate(request: Request, body: ApiRequest):
134
138
135
139
# Process click events to include full GitHub URLs
136
140
processed_diagram = process_click_events (
137
- mermaid_code ,
138
- body .username ,
139
- body .repo ,
140
- default_branch
141
+ mermaid_code , body .username , body .repo , default_branch
141
142
)
142
143
143
- return {"diagram" : processed_diagram ,
144
- "explanation" : explanation }
144
+ return {"diagram" : processed_diagram , "explanation" : explanation }
145
145
except RateLimitError as e :
146
146
raise HTTPException (
147
147
status_code = 429 ,
148
- detail = "Service is currently experiencing high demand. Please try again in a few minutes."
148
+ detail = "Service is currently experiencing high demand. Please try again in a few minutes." ,
149
149
)
150
150
except Exception as e :
151
151
return {"error" : str (e )}
@@ -184,12 +184,13 @@ def process_click_events(diagram: str, username: str, repo: str, branch: str) ->
184
184
Process click events in Mermaid diagram to include full GitHub URLs.
185
185
Detects if path is file or directory and uses appropriate URL format.
186
186
"""
187
+
187
188
def replace_path (match ):
188
189
# Extract the path from the click event
189
- path = match .group (2 ).strip ('" \' ' )
190
+ path = match .group (2 ).strip (" \" '" )
190
191
191
192
# Determine if path is likely a file (has extension) or directory
192
- is_file = '.' in path .split ('/' )[- 1 ]
193
+ is_file = "." in path .split ("/" )[- 1 ]
193
194
194
195
# Construct GitHub URL
195
196
base_url = f"https://github.com/{ username } /{ repo } "
0 commit comments