# -*- coding: utf-8 -*- """ rasa_agent.py — سرویس‌دهندهٔ رسا (Rasa DevOps Agent) · نسخهٔ 1.0.0 این فایل را روی هاست، کنارِ پوشهٔ rasa_backend بگذارید و به‌عنوان یک Python App (Application startup file = rasa_agent.py ، Entry point = application) اجرا کنید. استدیو از طریق این Agent می‌تواند وضعیت/نسخه را بخواند و (در فازهای بعد) مایگریشن اعمال کند. امنیت: هر درخواست باید فیلدِ secret برابرِ RASA_AGENT_SECRET داشته باشد. اتصالِ دیتابیس: از فایلِ .env کنارِ rasa_backend خوانده می‌شود (همان یوزر/رمزِ بک‌اند). """ import os import json import re AGENT_VERSION = "1.2.0" # کلیدِ امنیتی — استدیو موقعِ دانلود این را برایتان تزریق می‌کند؛ در صورت نیاز دستی هم می‌توانید عوض کنید. RASA_AGENT_SECRET = os.environ.get("RASA_AGENT_SECRET", "sRHg2EQXvjTqNsqsMtSwRJ06KWuKQrH5") _HERE = os.path.dirname(os.path.abspath(__file__)) # مسیرهای محتملِ .env (کنارِ خودِ agent یا داخلِ rasa_backend) _ENV_CANDIDATES = [ os.path.join(_HERE, ".env"), os.path.join(_HERE, "rasa_backend", ".env"), os.path.join(os.path.dirname(_HERE), "rasa_backend", ".env"), ] def _load_env(): cfg = {} for p in _ENV_CANDIDATES: if os.path.exists(p): try: with open(p, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) cfg[k.strip()] = v.strip() break except Exception: pass return cfg def _db(creds=None): """اتصالِ DB — creds (از فرمِ استقرار) بر .env اولویت دارد؛ برای نصبِ اولیه که هنوز .env نیست.""" import mysql.connector e = _load_env() c = creds or {} if c.get("db_user"): user, pw = c["db_user"], c.get("db_pass", "") else: user, pw = e.get("DB_USER", "root"), e.get("DB_PASS", "") # دیتابیسِ پیش‌فرض تعیین نمی‌شود؛ هر action در صورت نیاز خودش USE می‌کند # (وگرنه در نصبِ اولیه به DB_NAMEِ .envِ نامرتبط وصل می‌شد و خطای Unknown database می‌داد). return mysql.connector.connect( host=c.get("db_host") or e.get("DB_HOST", "localhost"), port=int(e.get("DB_PORT", "3306") or 3306), user=user, password=pw, database=c.get("database") or None, charset="utf8mb4", ) def _creds(body): return {k: body.get(k) for k in ("db_host", "db_user", "db_pass") if body.get(k)} # کوئریِ فقط‌خواندنی — فقط SELECT/SHOW/DESCRIBE؛ بدونِ چند-دستوری و کلماتِ خطرناک _READONLY = re.compile(r"^\s*(select|show|describe|desc|explain)\b", re.I) _FORBID = re.compile(r"\b(insert|update|delete|drop|alter|create|truncate|grant|replace|into\s+outfile)\b", re.I) def _handle(path, body): if path.endswith("/health"): return 200, {"ok": True, "agent": "rasa-agent", "version": AGENT_VERSION} # احراز if not body or body.get("secret") != RASA_AGENT_SECRET or RASA_AGENT_SECRET == "__AGENT_SECRET__": return 401, {"ok": False, "error": "کلیدِ Agent نامعتبر یا تنظیم‌نشده است"} action = (body.get("action") or "").strip() if action == "ping": return 200, {"ok": True, "version": AGENT_VERSION} if action == "db_info": e = _load_env() prefix = body.get("prefix") or e.get("DB_PREFIX", "") try: cn = _db(_creds(body)); cur = cn.cursor() cur.execute("SHOW DATABASES") dbs = [r[0] for r in cur.fetchall() if not prefix or r[0].startswith(prefix)] cn.close() return 200, {"ok": True, "databases": dbs, "prefix": prefix} except Exception as ex: return 500, {"ok": False, "error": str(ex)[:300]} if action == "query": sql = (body.get("sql") or "").strip().rstrip(";") if not _READONLY.match(sql) or _FORBID.search(sql) or ";" in sql: return 400, {"ok": False, "error": "فقط یک کوئریِ خواندنی (SELECT/SHOW) مجاز است"} try: cn = _db(_creds(body)); cur = cn.cursor() db = body.get("database") if db: cur.execute("USE `%s`" % re.sub(r"[^A-Za-z0-9_]", "", db)) cur.execute(sql) cols = [c[0] for c in cur.description] if cur.description else [] rows = [list(r) for r in cur.fetchall()[:1000]] cn.close() return 200, {"ok": True, "columns": cols, "rows": rows, "count": len(rows)} except Exception as ex: return 500, {"ok": False, "error": str(ex)[:300]} if action == "backup": dbs = body.get("databases") or [] if not dbs: return 400, {"ok": False, "error": "فهرستِ databases لازم است"} import subprocess import datetime e = _load_env() bdir = os.path.join(_HERE, "rasa_backups") try: os.makedirs(bdir, exist_ok=True) except Exception: pass ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") out = [] for db in dbs: db = re.sub(r"[^A-Za-z0-9_]", "", str(db)) if not db: continue fn = os.path.join(bdir, "%s_%s.sql" % (db, ts)) env = dict(os.environ); env["MYSQL_PWD"] = e.get("DB_PASS", "") try: with open(fn, "wb") as f: subprocess.run( ["mysqldump", "-h", e.get("DB_HOST", "localhost"), "-u", e.get("DB_USER", "root"), "--single-transaction", "--no-tablespaces", "--skip-lock-tables", "--routines", "--triggers", db], stdout=f, stderr=subprocess.DEVNULL, env=env, check=True, timeout=600) out.append({"db": db, "file": os.path.basename(fn), "size": os.path.getsize(fn)}) except Exception as ex: out.append({"db": db, "error": str(ex)[:200]}) ok = all("error" not in x for x in out) and bool(out) return (200 if ok else 500), {"ok": ok, "backups": out, "dir": bdir} if action == "migrate": stmts = body.get("statements") or [] database = body.get("database") allow_destructive = bool(body.get("allow_destructive")) if not stmts: return 400, {"ok": False, "error": "فهرستِ statements لازم است"} destruct = re.compile(r"\b(drop\s+database|drop\s+table|truncate)\b", re.I) results = [] try: cn = _db(_creds(body)); cn.autocommit = True; cur = cn.cursor() if database: cur.execute("USE `%s`" % re.sub(r"[^A-Za-z0-9_]", "", database)) for s in stmts: s2 = str(s).strip().rstrip(";") if not s2: continue if not allow_destructive and destruct.search(s2): results.append({"sql": s2[:90], "ok": False, "error": "مخرب — allow_destructive لازم است"}) continue try: cur.execute(s2) results.append({"sql": s2[:90], "ok": True, "rows": cur.rowcount}) except Exception as ex: results.append({"sql": s2[:90], "ok": False, "error": str(ex)[:200]}) cn.close() except Exception as ex: return 500, {"ok": False, "error": str(ex)[:300], "results": results} return 200, {"ok": all(r["ok"] for r in results), "results": results} if action == "write_env": path = body.get("path") content = body.get("content") if not path or content is None: return 400, {"ok": False, "error": "path و content لازم است"} try: d = os.path.dirname(path) if d and not os.path.exists(d): os.makedirs(d, exist_ok=True) with open(path, "w", encoding="utf-8", newline="\n") as f: f.write(content) return 200, {"ok": True, "path": path, "bytes": len(content)} except Exception as ex: return 500, {"ok": False, "error": str(ex)[:200]} if action == "extract_zip": import zipfile zpath = body.get("zip_path") dest = body.get("dest") if not zpath or not dest: return 400, {"ok": False, "error": "zip_path و dest لازم است"} try: os.makedirs(dest, exist_ok=True) with zipfile.ZipFile(zpath) as z: z.extractall(dest) n = len(z.namelist()) if body.get("delete_zip"): try: os.remove(zpath) except Exception: pass return 200, {"ok": True, "dest": dest, "files": n} except Exception as ex: return 500, {"ok": False, "error": str(ex)[:200]} return 400, {"ok": False, "error": "action نامعتبر"} def application(environ, start_response): """WSGI — سازگار با Passenger/cPanel.""" path = environ.get("PATH_INFO", "") or "" body = {} try: n = int(environ.get("CONTENT_LENGTH") or 0) if n: raw = environ["wsgi.input"].read(n) body = json.loads(raw.decode("utf-8") or "{}") except Exception: body = {} try: status_code, payload = _handle(path, body) except Exception as ex: status_code, payload = 500, {"ok": False, "error": str(ex)[:300]} out = json.dumps(payload, ensure_ascii=False).encode("utf-8") start_response("%d OK" % status_code, [ ("Content-Type", "application/json; charset=utf-8"), ("Content-Length", str(len(out))), ("Access-Control-Allow-Origin", "*"), ]) return [out] # اجرای مستقیم برای تست لوکال: python rasa_agent.py 9099 if __name__ == "__main__": import sys from wsgiref.simple_server import make_server port = int(sys.argv[1]) if len(sys.argv) > 1 else 9099 print("rasa-agent v%s on :%d" % (AGENT_VERSION, port)) make_server("0.0.0.0", port, application).serve_forever()