bamboo-cms · YogiLiu · Mar 11, 2024 · Mar 11, 2024 · Mar 12, 2024 · Apr 2, 2024
diff --git a/backend/bamboo/__init__.py b/backend/bamboo/__init__.py
@@ -1,7 +1,7 @@
 from apiflask import APIFlask
 from flask import current_app, send_from_directory
 
-from bamboo import blueprints, database, jobs
+from bamboo import blueprints, database, jobs, ssg
 from bamboo.settings import config
 
 
@@ -27,6 +27,8 @@ def create_app(config_name: str) -> APIFlask:
     database.init_app(app)
     # jobs
     jobs.init_app(app)
+    # ssg
+    ssg.init_app(app)
     # Serve media files for development environment.
     # This will be overriden by nginx in production environment.
     app.add_url_rule(f"{app.config['MEDIA_URL']}/<path:filename>", "media", media_endpoint)

diff --git a/backend/bamboo/jobs.py b/backend/bamboo/jobs.py
@@ -1,8 +1,16 @@
+import shutil
+import tempfile
+from io import BytesIO
 from pathlib import Path
+from zipfile import BadZipFile, ZipFile
 
 from flask import Flask, current_app
 from flask_rq2 import RQ
 from PIL import Image
+from sqlalchemy import ScalarResult
+
+from bamboo.database.models import Site, db
+from bamboo.utils import fetch_github_repo
 
 rq = RQ()
 
@@ -21,5 +29,41 @@ def gen_small_image(image_path: Path) -> None:
             small_image.save(small_image_path)
 
 
+@rq.job
+def sync_templates(store_dir: Path, **kwargs) -> None:
+    """
+    Sync the templates from the GitHub repository to local.
+
+    :param store_dir: The directory to store the template, every template will be store in a directory named {site.id}_{site.name} in this path.
+    :param gh_token: The GitHub token to use for the sync. If None, GitHub may be forbid access.
+    """
+    sites: ScalarResult[Site] = db.session.execute(db.select(Site)).scalars()
+    for site in sites:
+        if not site.template_url:
+            continue
+        name = f"{site.id}_{site.name}"
+        gh_token: str | None = kwargs.get("gh_token")
+        file_bytes = fetch_github_repo(site.template_url, gh_token=gh_token)
+        if file_bytes is None:
+            continue
+        buf = BytesIO(file_bytes)
+        try:
+            zip_file = ZipFile(buf)
+        except BadZipFile:
+            continue
+        if zip_file.testzip() is not None:
+            continue
+        if len(zip_file.namelist()) == 0:
+            continue
+        dir_name = zip_file.namelist()[0].split("/")[0]
+        dest = store_dir / name
+        if dest.exists():
+            shutil.rmtree(dest)
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            zip_file.extractall(tmp_dir)
+            shutil.copytree(Path(tmp_dir) / dir_name, dest)
+        zip_file.close()
+
+
 def init_app(app: Flask) -> None:
     rq.init_app(app)
diff --git a/backend/bamboo/ssg/__init__.py b/backend/bamboo/ssg/__init__.py
@@ -0,0 +1,15 @@
+from flask import Flask
+
+from bamboo.ssg.core import SSG
+
+
+def init_app(app: Flask, url_prefix: str = "/ssg", sync_interval: int = 5) -> None:
+    """
+    :param app: The Flask app to initialize with.
+    :param url_prefix: The URL prefix for the SSG blueprint.
+    :param sync_interval: The interval (in minutes) to sync the templates from GitHub.
+    """
+    ssg = SSG(app, url_prefix, sync_interval)
+    if not hasattr(app, "extensions"):
+        app.extensions = {}
+    app.extensions["ssg"] = ssg
diff --git a/backend/bamboo/ssg/core.py b/backend/bamboo/ssg/core.py
@@ -0,0 +1,81 @@
+from contextlib import contextmanager
+from pathlib import Path
+from tempfile import NamedTemporaryFile, TemporaryDirectory
+from typing import Generator
+from zipfile import ZipFile
+
+import jinja2
+from flask import Flask, abort
+
+from bamboo.database.models import Site
+from bamboo.jobs import sync_templates
+from bamboo.ssg.template import Environment, loader_func, url_filter_func
+from bamboo.ssg.views import freezer, site_bp
+
+static_dir = "static"  # JS, CSS, Image, etc. Send to browser directly.
+
+
+class SSG:
+    def __init__(self, app: Flask, url_prefix: str, sync_interval: int):
+        self.tpl_dir = Path(".") / "data" / "templates"  # 'data' in work dir
+        self.tpl_dir.mkdir(parents=True, exist_ok=True)
+        app.register_blueprint(site_bp, url_prefix=url_prefix)
+        gh_token = app.config.get("SSG_GH_TOKEN")
+        sync_templates.cron(
+            f"*/{sync_interval} * * * *",
+            args=(self.tpl_dir,),
+            kwargs={"gh_token": gh_token},
+            name="ssg",
+        )
+
+        self.jinja_env = Environment(loader=jinja2.FunctionLoader(loader_func(self.tpl_dir)))
+        self.jinja_env.filters["url"] = url_filter_func
+
+    def render(self, site: Site, tpl_file: str, **kwargs) -> bytes | str:
+        """
+        Render a template file for a site.
+
+        :param site: Site object, used for context data
+        :param tpl_file: Template file name
+        :param kwargs: Extra context data
+        """
+        if tpl_file.startswith(static_dir):
+            return self._send_static(site, tpl_file)
+        return self._render_template(site, tpl_file, **kwargs)
+
+    def _send_static(self, site: Site, file: str):
+        path = self.tpl_dir / f"{site.id}_{site.name}" / file
+        if not path.exists():
+            abort(404)
+        return path.read_bytes()
+
+    def _render_template(self, site: Site, tpl_file: str, **kwargs):
+        tpl_name = f"{site.id}_{site.name}/{tpl_file}"
+        try:
+            tpl = self.jinja_env.get_template(tpl_name)
+        except jinja2.TemplateNotFound:
+            abort(404)
+        return tpl.render(site=site, **kwargs)
+
+    @contextmanager
+    def pack(self, site: Site) -> Generator[Path, None, None]:
+        """
+        Pack all templates for a site into a zip file.
+        """
+        app = Flask("dummy", static_folder=None)
+        app.register_blueprint(site_bp)
+        app.config["SSG_PACKING"] = True
+        app.config["SSG_SITE"] = site
+        if not hasattr(app, "extensions"):
+            app.extensions = {}
+        app.extensions["ssg"] = self
+        with TemporaryDirectory() as tmpdir:
+            app.config["FREEZER_DESTINATION"] = str(tmpdir)
+            freezer.init_app(app)
+            freezer.freeze()
+            with NamedTemporaryFile("w+b", suffix=".zip") as tmpfile:
+                with ZipFile(tmpfile, "w") as zfile:
+                    for file in Path(tmpdir).rglob("*"):
+                        if file.is_file():
+                            zfile.write(file, file.relative_to(tmpdir))
+                yield Path(tmpfile.name)
diff --git a/backend/bamboo/ssg/template.py b/backend/bamboo/ssg/template.py
@@ -0,0 +1,34 @@
+from pathlib import Path
+
+import jinja2
+from flask import current_app
+
+from bamboo.database.models import Site
+
+
+def loader_func(base_dir: Path):
+    def loader(tpl_name: str) -> str:
+        return (base_dir / tpl_name).read_text()
+
+    return loader
+
+
+class Environment(jinja2.Environment):
+    def join_path(self, template: str, parent: str) -> str:
+        # if parent is not None, it is the name of the template that is including or importing the template.
+        if parent:
+            ps = parent.split("/", 1)
+            if len(ps) == 1:
+                return template
+            prefix = ps[0]
+            return f"{prefix}/{template}"
+        return template
+
+
+def url_filter_func(url: str, site: Site) -> str:
+    # if it packing site currently, do not add site_id
+    if current_app.config.get("SSG_PACKING"):
+        return url
+    if "?" in url:
+        return f"{url}&site_id={site.id}"
+    return f"{url}?site_id={site.id}"
diff --git a/backend/bamboo/ssg/views.py b/backend/bamboo/ssg/views.py
@@ -0,0 +1,63 @@
+from typing import TYPE_CHECKING
+
+from flask import Blueprint, current_app, g, request
+from flask_frozen import Freezer
+
+from bamboo.database import db
+from bamboo.database.models import Site
+
+if TYPE_CHECKING:
+    from bamboo.ssg.core import SSG
+
+
+# TODO: How to implement auth for this blueprint?
+# These pages canot not use token based auth.
+# Session / Cookies?
+site_bp = Blueprint("ssg", __name__)
+
+freezer = Freezer()
+
+
+@site_bp.before_request
+def validate_site_id():
+    if current_app.config.get("SSG_PACKING") and current_app.config.get("SSG_SITE"):
+        g.ssg_site = current_app.config.get("SSG_SITE")
+        return
+
+    site_id = request.args.get("site_id")
+    if not site_id:
+        return "site_id is required", 400
+    elif not site_id.isdigit():
+        return "invalid site_id", 400
+    site = db.get_or_404(Site, int(site_id))
+    g.ssg_site = site
+
+
+@site_bp.errorhandler(404)
+def page_not_found(error: Exception):
+    if g.ssg_site:
+        ssg: "SSG" = current_app.extensions["ssg"]
+        return ssg.render(g.ssg_site, "404.html")
+    return "resource not found", 404
+
+
+@site_bp.get("/static/<path:filename>")
+def static(filename: str):
+    ssg: "SSG" = current_app.extensions["ssg"]
+    return ssg.render(g.ssg_site, f"static/{filename}")
+
+
+@freezer.register_generator
+def static_gen():
+    if g.ssg_site:
+        ssg: "SSG" = current_app.extensions["ssg"]
+        path = ssg.tpl_dir / f"{g.ssg_site.id}_{g.ssg_site.name}" / "static"
+        for file in path.rglob("*"):
+            if file.is_file():
+                yield "ssg.static", {"filename": file.name}
+
+
+@site_bp.get("/")
+def index():
+    ssg: "SSG" = current_app.extensions["ssg"]
+    return ssg.render(g.ssg_site, "index.html")
diff --git a/backend/bamboo/utils.py b/backend/bamboo/utils.py
@@ -1,3 +1,4 @@
+import re
 import uuid
 from datetime import UTC, datetime, timedelta
 from typing import (
@@ -7,6 +8,7 @@
     MutableMapping,
 )
 
+import httpx
 from jose import jwt
 
 
@@ -104,3 +106,25 @@ def decode_jwt(
 def gen_uuid() -> str:
     """Generate a uuid hex string."""
     return str(uuid.uuid4().hex)
+
+
+gh_pattern = re.compile(r"github\.com/(?P<owner>\S+)/(?P<repo>\S+)")
+
+
+def fetch_github_repo(url: str, gh_token: str | None = None) -> bytes | None:
+    match = gh_pattern.search(url)
+    if match is None or "owner" not in match.groupdict() or "repo" not in match.groupdict():
+        return None
+    owner = match.group("owner")
+    repo = match.group("repo")
+    file_url = f"https://api.github.com/repos/{owner}/{repo}/zipball"
+    headers = {"Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28"}
+    if gh_token:
+        headers["Authorization"] = f"Bearer {gh_token}"
+    with httpx.Client() as client:
+        try:
+            res = client.get(file_url, headers=headers, follow_redirects=True, timeout=30)
+            res.raise_for_status()
+        except httpx.HTTPError:
+            return None
+        return res.content
diff --git a/backend/tests/ssg/__init__.py b/backend/tests/ssg/__init__.py
diff --git a/backend/tests/ssg/conftest.py b/backend/tests/ssg/conftest.py
@@ -0,0 +1,62 @@
+from io import BytesIO
+from pathlib import Path
+from zipfile import ZipFile
+
+import pytest
+from pytest_httpx import HTTPXMock
+
+from bamboo import create_app
+from bamboo.database import db
+from bamboo.database.models import Site
+
+
+@pytest.fixture
+def example_template_path():
+    return Path(__file__).parent / "template_example"
+
+
+@pytest.fixture
+def mock_template_download(httpx_mock: HTTPXMock, example_template_path: Path):
+    httpx_mock.add_response(
+        url="https://api.github.com/repos/PyConChina/templates/zipball",
+        status_code=302,
+        headers={
+            "Location": "https://codeload.github.com/PyConChina/templates/legacy.zip/refs/heads/main"
+        },
+    )
+    buf = BytesIO()
+    with ZipFile(buf, "w") as zip_file:
+        for file in example_template_path.rglob("*"):
+            if file.is_file():
+                p = Path("PyConChina-templates-0000000") / file.relative_to(example_template_path)
+                zip_file.write(file, p)
+    buf.seek(0)
+    httpx_mock.add_response(
+        url="https://codeload.github.com/PyConChina/templates/legacy.zip/refs/heads/main",
+        content=buf.read(),
+        headers={"Content-Type": "application/zip"},
+    )
+
+
+@pytest.fixture(autouse=True)
+def app():
+    app = create_app("testing")
+    with app.app_context():
+        yield app
+
+
+@pytest.fixture(autouse=True)
+def init_db():
+    db.create_all()
+    try:
+        yield
+    finally:
+        db.drop_all()
+
+
+@pytest.fixture(autouse=True)
+def site():
+    site = Site(id=1000, name="site1000", template_url="https://github.com/PyConChina/templates")
+    db.session.add(site)
+    db.session.commit()
+    yield site
diff --git a/backend/tests/ssg/template_example/index.html b/backend/tests/ssg/template_example/index.html
@@ -0,0 +1,7 @@
+{% extends "layout.html" %}
+{% block title %}{{ site.name }}{% endblock %}
+{% block body %}
+<div>index</div>
+<a href="{{ "test.html?a=1" | url(site) }}">Go</a>
+<script src="{{ "static/script.js" | url(site) }}"></script>
+{% endblock %}
diff --git a/backend/tests/ssg/template_example/layout.html b/backend/tests/ssg/template_example/layout.html
@@ -0,0 +1,15 @@
+<!DOCTYPE html>
+<html lang="zh">
+  <head>
+    <meta charset="UTF-8" />
+    <meta http-equiv="X-UA-Compatible" content="IE=edge" />
+    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+    <title>{% block title %}Welcome{% endblock %} | PyCon China</title>
+    {% block head %}
+      <link rel="stylesheet" href="{{ "static/style.css" | url(site) }}" />
+    {% endblock %}
+  </head>
+  <body>
+      <main>{% block body %}{% endblock %}</main>
+  </body>
+</html>