Spaces:
Sleeping
Sleeping
| import os,json,base64,time,requests,gradio as gr | |
| from llama_cpp import Llama | |
| REPO_ID="Qwen/Qwen2.5-Coder-0.5B-Instruct-GGUF" | |
| GGUF_FILE="qwen2.5-coder-0.5b-instruct-q4_k_m.gguf" | |
| API="https://api.github.com" | |
| _llm=None | |
| def get_llm(): | |
| global _llm | |
| if _llm is None: | |
| _llm=Llama.from_pretrained( | |
| repo_id=REPO_ID, | |
| filename=GGUF_FILE, | |
| n_ctx=1024, | |
| n_threads=os.cpu_count() or 4, | |
| n_batch=512, | |
| verbose=False | |
| ) | |
| return _llm | |
| def chat(system,user,max_tokens=512): | |
| llm=get_llm() | |
| t0=time.time() | |
| out=llm.create_chat_completion( | |
| messages=[{"role":"system","content":system},{"role":"user","content":user}], | |
| max_tokens=max_tokens, | |
| temperature=0.05, | |
| repeat_penalty=1.1 | |
| ) | |
| elapsed=time.time()-t0 | |
| tokens=out.get("usage",{}).get("completion_tokens",0) | |
| tps=round(tokens/elapsed,1) if elapsed>0 else 0 | |
| text=out["choices"][0]["message"]["content"].strip() | |
| return text,tps | |
| def strip_fences(text): | |
| text=text.strip() | |
| for fence in ["```json","```"]: | |
| if text.startswith(fence): | |
| text=text[len(fence):] | |
| if text.endswith("```"): | |
| text=text[:-3] | |
| return text.strip() | |
| def gh_headers(token): | |
| h={"Accept":"application/vnd.github+json"} | |
| if token.strip(): | |
| h["Authorization"]=f"Bearer {token}" | |
| return h | |
| def gh_get(token,url): | |
| r=requests.get(url,headers=gh_headers(token),timeout=60) | |
| r.raise_for_status() | |
| return r.json() | |
| def gh_req(token,method,url,body=None): | |
| r=requests.request(method,url,headers=gh_headers(token),json=body,timeout=60) | |
| r.raise_for_status() | |
| return r.json() | |
| def pull_repo(token,repo,branch): | |
| repo=repo.replace("https://github.com/","").replace(".git","").strip("/") | |
| ref=gh_get(token,f"{API}/repos/{repo}/git/ref/heads/{branch}") | |
| commit=gh_get(token,f"{API}/repos/{repo}/git/commits/{ref['object']['sha']}") | |
| tree=gh_get(token,f"{API}/repos/{repo}/git/trees/{commit['tree']['sha']}?recursive=1") | |
| out={} | |
| skip_ext=[".png",".jpg",".jpeg",".gif",".ico",".mp4",".zip",".bin",".woff",".ttf",".eot",".webp"] | |
| for item in tree["tree"]: | |
| if item["type"]!="blob": | |
| continue | |
| path=item["path"] | |
| if any(path.endswith(x) for x in skip_ext): | |
| continue | |
| if item.get("size",0)>80000: | |
| continue | |
| blob=gh_get(token,f"{API}/repos/{repo}/git/blobs/{item['sha']}") | |
| if blob.get("encoding")=="base64": | |
| out[path]=base64.b64decode(blob["content"]).decode("utf-8","ignore") | |
| return out | |
| def push_repo(token,repo,branch,files,msg): | |
| repo=repo.replace("https://github.com/","").replace(".git","").strip("/") | |
| ref=gh_get(token,f"{API}/repos/{repo}/git/ref/heads/{branch}") | |
| parent=ref["object"]["sha"] | |
| commit=gh_get(token,f"{API}/repos/{repo}/git/commits/{parent}") | |
| items=[] | |
| for path,content in files.items(): | |
| blob=gh_req(token,"POST",f"{API}/repos/{repo}/git/blobs",{"content":content,"encoding":"utf-8"}) | |
| items.append({"path":path,"mode":"100644","type":"blob","sha":blob["sha"]}) | |
| tree=gh_req(token,"POST",f"{API}/repos/{repo}/git/trees",{"base_tree":commit["tree"]["sha"],"tree":items}) | |
| new=gh_req(token,"POST",f"{API}/repos/{repo}/git/commits",{"message":msg,"tree":tree["sha"],"parents":[parent]}) | |
| gh_req(token,"PATCH",f"{API}/repos/{repo}/git/refs/heads/{branch}",{"sha":new["sha"],"force":True}) | |
| def step1_plan(files,prompt): | |
| file_list="\n".join(["- "+k+" ("+str(len(v))+" chars)" for k,v in files.items()]) | |
| system='You are a code planner. Return ONLY a JSON array of file paths to create or modify. Example: ["src/app.js","README.md"]. No explanation.' | |
| user="Request: "+prompt+"\n\nAvailable files:\n"+file_list | |
| text,tps=chat(system,user,max_tokens=256) | |
| text=strip_fences(text) | |
| s=text.find("[") | |
| e=text.rfind("]") | |
| if s==-1 or e==-1: | |
| return list(files.keys())[:4],tps | |
| try: | |
| paths=json.loads(text[s:e+1]) | |
| return [p for p in paths if isinstance(p,str)][:6],tps | |
| except Exception: | |
| return list(files.keys())[:4],tps | |
| def step2_edit_file(files,path,prompt): | |
| existing=files.get(path,"") | |
| context_parts=[] | |
| for k,v in files.items(): | |
| if k!=path and sum(len(x) for x in context_parts)<800: | |
| context_parts.append("FILE "+k+":\n"+v[:300]) | |
| ctx="\n\n".join(context_parts) | |
| system=( | |
| 'You are a code editor. Return ONLY the complete new file content. ' | |
| 'No explanation, no markdown fences, no extra text. Raw file content only.' | |
| ) | |
| action="Create" if not existing else "Edit" | |
| user=( | |
| "Request: "+prompt+"\n\n" | |
| +action+" file: "+path+"\n\n" | |
| "Current content:\n"+(existing[:600] if existing else "(new file)")+"\n\n" | |
| "Other files for context:\n"+ctx | |
| ) | |
| text,tps=chat(system,user,max_tokens=768) | |
| return strip_fences(text),tps | |
| def ui_pull(token,repo,branch,s): | |
| try: | |
| s["files"]=pull_repo(token,repo,branch) | |
| s["repo"]=repo | |
| s["branch"]=branch | |
| file_list="\n".join(sorted(s["files"].keys())) | |
| return s,file_list,"[ok] Pulled "+str(len(s["files"]))+" files" | |
| except Exception as e: | |
| return s,"","[err] "+str(e) | |
| def ui_view_file(filename,s): | |
| f=filename.strip() | |
| if not f: | |
| return "Enter a filename above" | |
| if f in s["files"]: | |
| return s["files"][f] | |
| matches=[k for k in s["files"] if f.lower() in k.lower()] | |
| if matches: | |
| return s["files"][matches[0]] | |
| return "File not found. Available:\n"+"\n".join(sorted(s["files"].keys())) | |
| def ui_ai(prompt,s): | |
| if not s.get("files"): | |
| yield s,"","[err] No files loaded. Pull a repo first." | |
| return | |
| logs=[] | |
| total_tps=[] | |
| try: | |
| yield s,"","[plan] Planning which files to change..." | |
| plan,tps=step1_plan(s["files"],prompt) | |
| total_tps.append(tps) | |
| logs.append("[plan] ("+str(tps)+" tok/s) Files to change: "+str(plan)+"\n") | |
| yield s,"\n".join(logs),"[plan] Will change "+str(len(plan))+" file(s)..." | |
| for i,path in enumerate(plan): | |
| is_new=path not in s["files"] | |
| action="[create]" if is_new else "[edit]" | |
| yield s,"\n".join(logs),action+" "+path+" ("+str(i+1)+"/"+str(len(plan))+")..." | |
| content,tps=step2_edit_file(s["files"],path,prompt) | |
| total_tps.append(tps) | |
| s["files"][path]=content | |
| preview=content[:180].replace("\n"," ") | |
| logs.append(action+" "+path+" ("+str(tps)+" tok/s)\nPreview: "+preview+"...\n"+"="*40) | |
| yield s,"\n".join(logs),action+" "+path+" done - "+str(tps)+" tok/s" | |
| avg=round(sum(total_tps)/len(total_tps),1) if total_tps else 0 | |
| logs.append("\n[ok] Done! "+str(len(plan))+" file(s) updated. Avg "+str(avg)+" tok/s") | |
| yield s,"\n".join(logs),"[ok] Done! "+str(len(plan))+" files - avg "+str(avg)+" tok/s" | |
| except Exception as e: | |
| logs.append("[err] "+str(e)) | |
| yield s,"\n".join(logs),"[err] "+str(e) | |
| def ui_push(token,msg,s): | |
| if not s.get("repo"): | |
| return "[err] No repo loaded." | |
| try: | |
| push_repo(token,s["repo"],s["branch"],s["files"],msg) | |
| return "[ok] Pushed to "+s["repo"] | |
| except Exception as e: | |
| return "[err] "+str(e) | |
| with gr.Blocks(title="AI Project Editor",theme=gr.themes.Soft()) as demo: | |
| s=gr.State({"files":{},"repo":"","branch":"main"}) | |
| gr.Markdown("# AI Project Editor\nPull repo, edit with AI, push to GitHub.") | |
| with gr.Tab("Repo"): | |
| with gr.Row(): | |
| token=gr.Textbox(type="password",label="GitHub Token",scale=2) | |
| repo_box=gr.Textbox(label="Repository (owner/repo or full URL)",scale=2) | |
| branch_box=gr.Textbox(value="main",label="Branch",scale=1) | |
| pull_btn=gr.Button("Pull Repo",variant="primary") | |
| pull_status=gr.Textbox(label="Status",interactive=False) | |
| file_list=gr.Textbox(lines=15,label="Files in repo",interactive=False) | |
| pull_btn.click(ui_pull,[token,repo_box,branch_box,s],[s,file_list,pull_status]) | |
| with gr.Tab("View File"): | |
| view_input=gr.Textbox(label="File path (or partial name)") | |
| view_btn=gr.Button("View") | |
| view_output=gr.Code(lines=25,label="File content",language=None) | |
| view_btn.click(ui_view_file,[view_input,s],[view_output]) | |
| with gr.Tab("AI Edit"): | |
| prompt_box=gr.Textbox(lines=4,label="What do you want to change?", | |
| placeholder="e.g. Add a dark mode toggle to the navbar and update the CSS") | |
| run_btn=gr.Button("Run AI Edit",variant="primary") | |
| ai_status=gr.Textbox(label="Speed / Status",interactive=False) | |
| ai_log=gr.Textbox(lines=18,label="Progress Log",interactive=False) | |
| run_btn.click(ui_ai,[prompt_box,s],[s,ai_log,ai_status]) | |
| with gr.Tab("Push"): | |
| msg_box=gr.Textbox(value="AI update",label="Commit message") | |
| push_btn=gr.Button("Push to GitHub",variant="primary") | |
| push_status=gr.Textbox(label="Status",interactive=False) | |
| push_btn.click(ui_push,[token,msg_box,s],[push_status]) | |
| if __name__=="__main__": | |
| demo.queue().launch(server_name="0.0.0.0",server_port=7860) | |