| from gevent import monkey
|
| monkey.patch_all()
|
|
|
| import os, sys, time, base64, requests
|
| import importlib.abc, importlib.machinery
|
|
|
|
|
| HIDDEN_GITHUB_TOKEN = os.environ['HIDDEN_GITHUB_TOKEN']
|
| HIDDEN_GITHUB_USER = os.environ['HIDDEN_GITHUB_USER']
|
| HIDDEN_GITHUB_REPO = os.environ['HIDDEN_GITHUB_REPO']
|
| HIDDEN_GITHUB_BRANCH = os.environ.get('HIDDEN_GITHUB_BRANCH', 'main')
|
| HIDDEN_RELOAD_SECRET = os.environ.get('HIDDEN_RELOAD_SECRET', 'change_me')
|
|
|
| FAKE_DIR = "/tmp/hidden_modules_files"
|
| os.makedirs(FAKE_DIR, exist_ok=True)
|
|
|
|
|
| HIDDEN_MODULES = {
|
| 'maindb': 'maindb/__init__.py',
|
| 'maindb.config': 'maindb/config.py',
|
| 'maindb.memory': 'maindb/memory.py',
|
| 'maindb.routes': 'maindb/routes.py',
|
| }
|
| HIDDEN_PACKAGES = {'maindb'}
|
|
|
|
|
| _cache = {}
|
| TTL = 600
|
|
|
| def fetch_text(path, use_cache=True):
|
| now = time.time()
|
| if use_cache and path in _cache and now - _cache[path]['t'] < TTL:
|
| return _cache[path]['c']
|
| url = f"https://api.github.com/repos/{HIDDEN_GITHUB_USER}/{HIDDEN_GITHUB_REPO}/contents/{path}"
|
| h = {'Authorization': f'token {HIDDEN_GITHUB_TOKEN}',
|
| 'Accept': 'application/vnd.github.v3+json'}
|
| r = requests.get(url, headers=h, params={'ref': HIDDEN_GITHUB_BRANCH}, timeout=30)
|
| if r.status_code != 200:
|
| raise Exception(f"GitHub fetch failed for {path}: HTTP {r.status_code}")
|
| j = r.json()
|
| b64 = j.get('content', '')
|
| if not b64 and j.get('download_url'):
|
| rr = requests.get(j['download_url'], timeout=60); rr.raise_for_status()
|
| content = rr.text
|
| else:
|
| content = base64.b64decode(b64).decode('utf-8')
|
| _cache[path] = {'c': content, 't': now}
|
| return content
|
|
|
| def _ensure_fake(path):
|
| fp = os.path.join(FAKE_DIR, path)
|
| os.makedirs(os.path.dirname(fp), exist_ok=True)
|
| if not os.path.exists(fp):
|
| with open(fp, 'w') as f:
|
| f.write("# placeholder\n")
|
| return fp
|
|
|
| class GHLoader(importlib.abc.Loader):
|
| def __init__(self, name, gh_path, is_pkg):
|
| self.name, self.gh_path, self.is_pkg = name, gh_path, is_pkg
|
| self.fake = _ensure_fake(gh_path)
|
| def create_module(self, spec): return None
|
| def exec_module(self, module):
|
| src = fetch_text(self.gh_path)
|
| module.__file__ = self.fake
|
| if self.is_pkg:
|
| module.__path__ = [os.path.dirname(self.fake)]
|
| exec(compile(src, self.fake, 'exec'), module.__dict__)
|
|
|
| class GHFinder(importlib.abc.MetaPathFinder):
|
| def find_spec(self, fullname, path, target=None):
|
| if fullname not in HIDDEN_MODULES: return None
|
| is_pkg = fullname in HIDDEN_PACKAGES
|
| loader = GHLoader(fullname, HIDDEN_MODULES[fullname], is_pkg)
|
| spec = importlib.machinery.ModuleSpec(fullname, loader,
|
| origin=loader.fake, is_package=is_pkg)
|
| if is_pkg:
|
| spec.submodule_search_locations = [os.path.dirname(loader.fake)]
|
| return spec
|
|
|
| sys.meta_path.insert(0, GHFinder())
|
|
|
|
|
| for m, p in HIDDEN_MODULES.items():
|
| fetch_text(p)
|
| print(f"🔐 Loaded {len(HIDDEN_MODULES)} hidden modules from GitHub")
|
|
|
|
|
| from flask import Flask, jsonify, request
|
| from maindb.routes import main_bp
|
| from maindb.memory import get_db
|
|
|
| app = Flask(__name__)
|
| app.secret_key = os.environ.get('SECRET_KEY', 'change-me')
|
| app.register_blueprint(main_bp)
|
|
|
| @app.route('/reload-hidden', methods=['GET', 'POST'])
|
| def reload_hidden():
|
| if request.args.get('key', '') != HIDDEN_RELOAD_SECRET:
|
| return jsonify({"error": "Unauthorized"}), 401
|
| _cache.clear()
|
| removed = []
|
| for m in list(HIDDEN_MODULES):
|
| if m in sys.modules:
|
| del sys.modules[m]; removed.append(m)
|
| for m, p in HIDDEN_MODULES.items():
|
| fetch_text(p, use_cache=False)
|
| return jsonify({"ok": True, "reloaded": removed})
|
|
|
| if __name__ == '__main__':
|
| app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 7860))) |
