Spaces:
Running
Running
| import os | |
| from flask import Flask, request, jsonify, Response, stream_with_context, send_file, render_template | |
| from flask_cors import CORS | |
| import json, time, re, io, base64, html, logging | |
| import requests | |
| from urllib.parse import quote_plus | |
| from g4f.client import ClientFactory | |
| from docx import Document | |
| from pptx import Presentation | |
| from openpyxl import Workbook | |
| from reportlab.lib.pagesizes import A4 | |
| from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle | |
| from reportlab.lib.styles import getSampleStyleSheet | |
| from reportlab.lib import colors | |
| try: | |
| import PyPDF2 | |
| except: | |
| PyPDF2 = None | |
| # ======================== Logging ======================== | |
| logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S") | |
| logger = logging.getLogger("genisi") | |
| # ======================== Base ======================== | |
| BASE_DIR = os.path.dirname(os.path.abspath(__file__)) | |
| TEMPLATE_DIR = os.path.join(BASE_DIR, 'templates') | |
| app = Flask(__name__, template_folder=TEMPLATE_DIR) | |
| CORS(app) | |
| if not os.path.isfile(os.path.join(TEMPLATE_DIR, 'index.html')): | |
| raise RuntimeError("index.html not found in templates/") | |
| # ======================== API Configuration ======================== | |
| G4F_API_KEY = "g4f_u_mr2i6a_d57da5201b897515fe32b881e8c579440a6dbe70df88f574_da069b53" | |
| # ✨ Gemma Provider | |
| GEMMA_PROVIDER = "custom:srv_mrgy0nmbc8a86c407f17" | |
| GEMMA_MODEL = "models/gemma-4-31b-it" | |
| # عميل Gemma | |
| GEMMA_CLIENT = ClientFactory.create_client(GEMMA_PROVIDER, api_key=G4F_API_KEY) | |
| # وسائط | |
| MEDIA_CLIENT = ClientFactory.create_client("custom:srv_mp5miql908c8738d71be", api_key=G4F_API_KEY) | |
| IMAGE_MODEL = "flux" | |
| CAPS = { | |
| "flash": {"name": "Genisi Flash 2 ⚡", "vision": False, "imageMimes": [], "docExts": ["pdf","docx","txt","md","csv","json","js","ts","py","html","css","java","c","cpp"]}, | |
| "pro": {"name": "Genisi Pro 2 🧠", "vision": False, "imageMimes": [], "docExts": ["pdf","docx","txt","md","csv","json","js","ts","py","html","css","java","c","cpp"]}, | |
| } | |
| IMAGE_LIMITS = {} | |
| IMG_WINDOW = 3 * 3600 | |
| IMG_LIMIT = 3 | |
| def check_limit(store, client_id, window, limit): | |
| now = time.time() | |
| hist = [ts for ts in store.get(client_id, []) if now - ts < window] | |
| store[client_id] = hist | |
| return hist, now | |
| # ======================== Routes ======================== | |
| def index(): | |
| return render_template('index.html') | |
| def favicon(): | |
| return Response('', status=204) | |
| def health(): | |
| return jsonify({ | |
| 'status': 'ok', | |
| 'project': 'Genisi AI', | |
| 'owner': 'Anes Kameche (أنس كامش)', | |
| 'developer': 'AnesNT Project', | |
| 'provider': f'Gemma ({GEMMA_PROVIDER})', | |
| 'model': GEMMA_MODEL, | |
| 'models': { | |
| 'flash': 'Genisi Flash 2 ⚡ - Gemma 4 31B', | |
| 'pro': 'Genisi Pro 2 🧠 - Gemma 4 31B (Deep Thinking)' | |
| } | |
| }) | |
| # ======================== System Prompts ======================== | |
| FLASH_PROMPT = """You are Genisi Flash 2 ⚡, a fast and direct AI assistant, developed by the AnesNT project, owned by Anes Kameche (أنس كامش). | |
| CRITICAL RULES: | |
| 1. NO THINKING - ANSWER IMMEDIATELY. Never start with "Thinking..." or any reasoning. | |
| 2. Answer in the SAME language the user writes in. | |
| 3. Keep responses SHORT and DIRECT. | |
| 4. Use simple, clear language. | |
| TOOLS (output ONE JSON object when user explicitly requests): | |
| - Web search: {"tool":"search","query":"search terms"} | |
| - Image search: {"tool":"image_search","query":"search terms"} | |
| - Image generation: {"tool":"image","prompt":"detailed English description"} | |
| - Document: {"tool":"document","format":"pdf|docx|pptx|xlsx","title":"Document title","content":{...}} | |
| END EVERY RESPONSE WITH: | |
| SUGGESTIONS: ["suggestion 1","suggestion 2","suggestion 3"] | |
| (2-4 short follow-up prompts in the user's language)""" | |
| PRO_PROMPT = """You are Genisi Pro 2 🧠, a highly analytical and precise AI assistant powered, developed by the AnesNT project, owned by Anes Kameche (أنس كامش). | |
| CRITICAL RULES: | |
| 1. MANDATORY DEEP THINKING: Start EVERY response with "Thinking..." then your complete internal reasoning. After reasoning, add a blank line, then your final answer. | |
| 2. ANSWER MUST BE: Comprehensive, Accurate, Structured, Informative, Honest, Balanced, Actionable. | |
| 3. NEVER fabricate facts. ALWAYS verify before stating. Use web search for real-time info. | |
| 4. Use professional formatting: ## headings, **bold**, bullet points, tables. | |
| 5. TOOLS (output ONE JSON object when needed): | |
| - Web search: {"tool":"search","query":"search terms"} | |
| - Image search: {"tool":"image_search","query":"search terms"} | |
| - Image generation: {"tool":"image","prompt":"detailed English description"} | |
| - Document: {"tool":"document","format":"pdf|docx|pptx|xlsx","title":"Document title","content":{...}} | |
| END EVERY RESPONSE WITH: | |
| SUGGESTIONS: ["suggestion 1","suggestion 2","suggestion 3"] | |
| (2-4 insightful follow-up prompts in the user's language)""" | |
| def build_prompt(user, interests, think): | |
| if think: | |
| lines = [PRO_PROMPT] | |
| else: | |
| lines = [FLASH_PROMPT] | |
| user = user or {} | |
| if user.get('name'): | |
| lines.append(f"\nCurrent user: {user['name']}") | |
| if interests: | |
| lines.append(f"User interests: {', '.join(interests[:10])}") | |
| return "\n".join(lines) | |
| # ======================== Helpers ======================== | |
| def as_text(v): | |
| if v is None: return "" | |
| if isinstance(v, str): return v | |
| if isinstance(v, list): return "\n".join(as_text(x) for x in v) | |
| if isinstance(v, dict): return v.get('text') or v.get('body') or v.get('content') or "" | |
| return str(v) | |
| def normalize_messages(messages): | |
| out = [] | |
| for m in messages[-20:]: | |
| role = m.get('role') | |
| content = m.get('content', '') or '' | |
| out.append({"role": role, "content": content}) | |
| return out | |
| def extract_media_url(text): | |
| if not text: return None | |
| m = re.search(r'!\[.*?\]\((.*?)\)', text) or re.search(r'(https?://\S+)', text) | |
| return m.group(1).rstrip(').,') if m else None | |
| # ======================== Search ======================== | |
| search_cache = {} | |
| def web_search(query, n=5): | |
| key = query.lower().strip() | |
| if key in search_cache and time.time() - search_cache[key][0] < 300: | |
| return search_cache[key][1] | |
| try: | |
| r = requests.post("https://html.duckduckgo.com/html/", data={"q": query}, headers={"User-Agent": "Mozilla/5.0"}, timeout=3) | |
| results = [] | |
| for m in re.finditer(r'<a rel="nofollow" class="result__a" href="(.*?)".*?>(.*?)</a>.*?class="result__snippet".*?>(.*?)</a>', r.text, re.S): | |
| results.append({"title": html.unescape(re.sub('<.*?>', '', m.group(2))).strip(), "url": html.unescape(re.sub('<.*?>', '', m.group(1))).strip(), "snippet": html.unescape(re.sub('<.*?>', '', m.group(3))).strip()}) | |
| if len(results) >= n: break | |
| if results: search_cache[key] = (time.time(), results) | |
| return results | |
| except: return [] | |
| def image_search(query, n=6): | |
| try: | |
| r = requests.get("https://www.bing.com/images/search", params={"q": query}, headers={"User-Agent": "Mozilla/5.0"}, timeout=3) | |
| return [{"image": u} for u in re.findall(r'murl":"(.*?)"', r.text)[:n]] | |
| except: return [] | |
| # ======================== Documents ======================== | |
| def build_docx(cmd): | |
| doc = Document() | |
| doc.add_heading(cmd.get('title', 'Document'), 0) | |
| c = cmd.get('content', {}) or {} | |
| if c.get('heading'): doc.add_heading(c['heading'], 1) | |
| for p in c.get('paragraphs', []): doc.add_paragraph(as_text(p)) | |
| for s in c.get('sections', []): | |
| if s.get('title'): doc.add_heading(s['title'], 2) | |
| if s.get('body'): doc.add_paragraph(as_text(s['body'])) | |
| for b in s.get('bullets', []): doc.add_paragraph(as_text(b), style='List Bullet') | |
| if c.get('headers'): | |
| t = doc.add_table(rows=1, cols=len(c['headers'])) | |
| for i, h in enumerate(c['headers']): t.rows[0].cells[i].text = as_text(h) | |
| for r in c.get('rows', []): | |
| cells = t.add_row().cells | |
| for i, v in enumerate(r): | |
| if i < len(cells): cells[i].text = as_text(v) | |
| buf = io.BytesIO(); doc.save(buf); buf.seek(0); return buf | |
| def build_pptx(cmd): | |
| prs = Presentation() | |
| prs.slides.add_slide(prs.slide_layouts[0]).shapes.title.text = cmd.get('title', 'Presentation') | |
| for s in (cmd.get('content', {}).get('slides') or []): | |
| sl = prs.slides.add_slide(prs.slide_layouts[1]) | |
| sl.shapes.title.text = s.get('title', '') | |
| b = s.get('bullets', []) | |
| if b: | |
| tf = sl.placeholders[1].text_frame; tf.text = as_text(b[0]) | |
| for x in b[1:]: tf.add_paragraph().text = as_text(x) | |
| buf = io.BytesIO(); prs.save(buf); buf.seek(0); return buf | |
| def build_xlsx(cmd): | |
| wb = Workbook(); ws = wb.active; ws.title = (cmd.get('title') or 'Sheet')[:31] | |
| c = cmd.get('content', {}) or {} | |
| if c.get('headers'): ws.append([as_text(h) for h in c['headers']]) | |
| for r in c.get('rows', []): ws.append([as_text(v) for v in r]) | |
| buf = io.BytesIO(); wb.save(buf); buf.seek(0); return buf | |
| def build_pdf(cmd): | |
| buf = io.BytesIO() | |
| doc = SimpleDocTemplate(buf, pagesize=A4) | |
| styles = getSampleStyleSheet(); story = [] | |
| story.append(Paragraph(cmd.get('title', 'Document'), styles['Title'])); story.append(Spacer(1, 12)) | |
| c = cmd.get('content', {}) or {} | |
| if c.get('heading'): story.append(Paragraph(c['heading'], styles['Heading1'])) | |
| for p in c.get('paragraphs', []): story.append(Paragraph(as_text(p), styles['Normal'])) | |
| if c.get('headers'): | |
| data = [c['headers']] + c.get('rows', []) | |
| tbl = Table([[as_text(v) for v in row] for row in data]) | |
| tbl.setStyle(TableStyle([('BACKGROUND', (0, 0), (-1, 0), colors.HexColor('#7c3aed')), ('TEXTCOLOR', (0, 0), (-1, 0), colors.white), ('GRID', (0, 0), (-1, -1), 0.5, colors.grey)])) | |
| story.append(tbl) | |
| doc.build(story); buf.seek(0); return buf | |
| def extract_text(file_storage, ext): | |
| data = file_storage.read() | |
| if ext == 'pdf' and PyPDF2: | |
| try: return "\n".join((p.extract_text() or "") for p in PyPDF2.PdfReader(io.BytesIO(data)).pages) | |
| except: pass | |
| if ext == 'docx': | |
| try: return "\n".join(p.text for p in Document(io.BytesIO(data)).paragraphs) | |
| except: pass | |
| try: return data.decode('utf-8', errors='ignore') | |
| except: return "" | |
| # ======================== Tools ======================== | |
| def extract_tool(text): | |
| try: | |
| m = re.compile(r'\{[^{}]*"tool"\s*:\s*"[^"]+"\s*[^{}]*\}', re.DOTALL).search(text) | |
| return json.loads(m.group()) if m else None | |
| except: return None | |
| def execute_tool(td): | |
| t = td.get('tool') | |
| if t == 'search': | |
| r = web_search(td.get('query', ''), 5) | |
| return "\n\n### 🔍 Search:\n\n" + "\n\n".join(f"**{i+1}. {x['title']}**\n{x['snippet'][:150]}\n{x['url']}" for i, x in enumerate(r)) if r else "\n\n⚠️ No results.\n\n" | |
| elif t == 'image_search': | |
| imgs = image_search(td.get('query', '')) | |
| return "\n\n### 🖼️:\n\n" + "\n\n".join(f"" for x in imgs[:4]) if imgs else "\n\n⚠️ No images.\n\n" | |
| elif t == 'image': | |
| return {"special": "image", "prompt": td.get('prompt', '')} | |
| elif t == 'document': | |
| return "\n\n📄 Creating...\n\n" | |
| return None | |
| # ======================== Chat ======================== | |
| def _safe_json(): | |
| try: return request.get_json(force=True, silent=True) or {} | |
| except: return {} | |
| def chat(): | |
| try: | |
| data = _safe_json() | |
| if not data: return jsonify({'error': 'Empty'}), 400 | |
| messages = data.get('messages', []) | |
| model_type = data.get('model', 'flash') | |
| user = data.get('user') or {} | |
| interests = data.get('interests') or [] | |
| think = model_type == 'pro' | |
| last_msg = messages[-1].get('content', '') if messages else '' | |
| sm = re.search(r'(ابحث|بحث|search|find|معلومات عن|اخبار|من هو|ما هي|ما هو)\s+(.+)', last_msg, re.I) | |
| if sm: | |
| results = web_search(sm.group(2).strip(), 5) | |
| if results: | |
| ctx = "\n\n📚 Web:\n\n" + "\n".join(f"- {r['title']}: {r['snippet'][:200]}\n {r['url']}" for r in results) | |
| messages[-1]['content'] = f"Use this info:\n{ctx}\n\nQuestion: {last_msg}" | |
| sp = build_prompt(user, interests, think) | |
| full_msgs = [{"role": "system", "content": sp}] + normalize_messages(messages[-10:]) | |
| def generate(): | |
| full = "" | |
| try: | |
| name = "Genisi Pro 2 🧠" if think else "Genisi Flash 2 ⚡" | |
| logger.info(f"[chat] {name} | Gemma 4 31B") | |
| response = GEMMA_CLIENT.chat.completions.create(model=GEMMA_MODEL, messages=full_msgs, stream=True) | |
| for chunk in response: | |
| delta = getattr(chunk.choices[0].delta, 'content', None) | |
| if delta: | |
| full += delta | |
| yield f"data: {json.dumps({'delta': delta})}\n\n" | |
| tool = extract_tool(full) | |
| if tool: | |
| logger.info(f"[tool] {tool.get('tool')}") | |
| tr = execute_tool(tool) | |
| if isinstance(tr, dict) and tr.get('special') == 'image': | |
| yield f"data: {json.dumps({'tool': 'image', 'prompt': tr['prompt']})}\n\n" | |
| elif tr: | |
| yield f"data: {json.dumps({'delta': tr})}\n\n" | |
| yield f"data: {json.dumps({'done': True})}\n\n" | |
| except Exception as e: | |
| logger.error(f"[chat] {e}") | |
| yield f"data: {json.dumps({'error': f'Error: {str(e)[:200]}'})}\n\n" | |
| yield f"data: {json.dumps({'done': True})}\n\n" | |
| return Response(stream_with_context(generate()), mimetype='text/event-stream') | |
| except Exception as e: | |
| return jsonify({'error': str(e)}), 500 | |
| def title(): | |
| try: | |
| data = _safe_json() | |
| convo = "\n".join(f"{m.get('role')}: {(m.get('content') or '')[:300]}" for m in data.get('messages', [])[:4]) | |
| resp = GEMMA_CLIENT.chat.completions.create(model=GEMMA_MODEL, messages=[{"role": "user", "content": f"Short title (max 5 words):\n{convo}"}], stream=False) | |
| txt = (resp.choices[0].message.content or "").strip().strip('"').split("\n")[0][:60] | |
| return jsonify({'title': txt or None}) | |
| except Exception as e: | |
| return jsonify({'title': None}) | |
| def upload(): | |
| f = request.files.get('file') | |
| if not f: return jsonify({'message': 'No file'}), 400 | |
| name, ext = f.filename or 'file', (f.filename or 'file').rsplit('.', 1)[-1].lower() if '.' in (f.filename or '') else '' | |
| caps = CAPS['flash'] | |
| if f.mimetype in caps['imageMimes']: | |
| b64 = base64.b64encode(f.read()).decode() | |
| return jsonify({'kind': 'image', 'name': name, 'dataUrl': f'data:{f.mimetype};base64,{b64}'}) | |
| if ext not in caps['docExts']: return jsonify({'message': 'Unsupported'}), 400 | |
| return jsonify({'kind': 'text', 'name': name, 'text': extract_text(f, ext)[:20000]}) | |
| def image_gen(): | |
| data = _safe_json() | |
| prompt, cid = data.get('prompt', ''), data.get('clientId', 'anon') | |
| hist, now = check_limit(IMAGE_LIMITS, cid, IMG_WINDOW, IMG_LIMIT) | |
| if len(hist) >= IMG_LIMIT: return jsonify({'message': 'Limit', 'remaining': 0}) | |
| try: | |
| resp = MEDIA_CLIENT.chat.completions.create(model=IMAGE_MODEL, messages=[{"role": "user", "content": prompt}]) | |
| url = extract_media_url(resp.choices[0].message.content or "") | |
| if not url: return jsonify({'message': 'Failed'}) | |
| r = requests.get(url, timeout=60, headers={'User-Agent': 'Mozilla/5.0'}) | |
| b64 = base64.b64encode(r.content).decode() | |
| hist.append(now); IMAGE_LIMITS[cid] = hist | |
| return jsonify({'image': f"data:{r.headers.get('Content-Type', 'image/png')};base64,{b64}", 'remaining': IMG_LIMIT - len(hist)}) | |
| except Exception as e: | |
| return jsonify({'message': str(e)}) | |
| def document(): | |
| try: | |
| cmd = _safe_json() | |
| fmt = (cmd.get('format') or 'pdf').lower() | |
| builders = { | |
| 'docx': (build_docx, 'application/vnd.openxmlformats-officedocument.wordprocessingml.document'), | |
| 'pptx': (build_pptx, 'application/vnd.openxmlformats-officedocument.presentationml.presentation'), | |
| 'xlsx': (build_xlsx, 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'), | |
| 'pdf': (build_pdf, 'application/pdf'), | |
| } | |
| b, m = builders.get(fmt, (build_pdf, 'application/pdf')) | |
| buf = b(cmd) | |
| return send_file(buf, mimetype=m, as_attachment=True, download_name=f"{cmd.get('title', 'doc')}.{fmt}") | |
| except Exception as e: | |
| return jsonify({'error': str(e)}), 500 | |
| def search_ep(): | |
| q = _safe_json().get('query', '') | |
| return jsonify({'results': web_search(q) if q else []}) | |
| def image_search_ep(): | |
| q = _safe_json().get('query', '') | |
| return jsonify({'images': image_search(q) if q else []}) | |
| def img_proxy(): | |
| url = request.args.get('u') | |
| if not url: return '', 400 | |
| try: | |
| r = requests.get(url, timeout=30, stream=True, headers={'User-Agent': 'Mozilla/5.0'}) | |
| return Response(r.iter_content(8192), content_type=r.headers.get('Content-Type', 'application/octet-stream')) | |
| except: return '', 502 | |
| def nf(e): return jsonify({'error': 'Not found'}), 404 | |
| if __name__ == '__main__': | |
| port = int(os.environ.get('PORT', 7860)) | |
| logger.info("=" * 55) | |
| logger.info(" 🤖 Genisi AI Server") | |
| logger.info(" Owner: Anes Kameche (أنس كامش)") | |
| logger.info(" Developer: AnesNT Project") | |
| logger.info(f" Provider: Gemma ({GEMMA_PROVIDER})") | |
| logger.info(f" Model: {GEMMA_MODEL}") | |
| logger.info(" Flash ⚡ | Pro 🧠 (Deep Thinking)") | |
| logger.info(f" Port: {port}") | |
| logger.info("=" * 55) | |
| app.run(host='0.0.0.0', port=port, threaded=True) |