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)