feat: API for installation

server/model/schema.py

@@ -129,6 +129,13 @@ class BindUser(Base):
     email = db.Column(db.String(128), nullable=True, comment="邮箱")
     name = db.Column(db.String(128), nullable=True, comment="用户名")
     avatar = db.Column(db.String(256), nullable=True, comment="头像")
+    access_token = db.Column(
+        db.String(128), nullable=True, comment="GitHub access_token"
+    )
+    refresh_token = db.Column(
+        db.String(128), nullable=True, comment="GitHub refresh_token"
+    )
+    expire_time = db.Column(db.Integer, nullable=True, comment="GitHub token过期时间 时间戳")
     extra = db.Column(
         JSONStr(2048), nullable=True, server_default=text("'{}'"), comment="用户其他字段"
     )

server/model/team.py

@@ -1,4 +1,4 @@
-from flask import abort
+from flask import abort, session
 from sqlalchemy import and_, or_
 from utils.utils import query_one_page
 
@@ -122,6 +122,76 @@ def set_team_member(team_id, code_user_id, im_user_id):
     db.session.commit()
 
 
+def create_team(app_info: dict) -> Team:
+    """Create a team.
+
+    Args:
+        name (str): Team name.
+        app_info (dict): GitHub App info.
+
+    Returns:
+        Team: Team object.
+    """
+
+    current_user_id = session.get("user_id", None)
+    if not current_user_id:
+        abort(403, "can not found user by id")
+
+    new_team = Team(
+        id=ObjID.new_id(),
+        user_id=current_user_id,
+        name=app_info["account"]["login"],
+        description=None,
+        # extra=app_info,
+    )
+
+    db.session.add(new_team)
+    db.session.flush()
+
+    # 创建 TeamMember
+    current_bind_user = BindUser.query.filter(
+        BindUser.user_id == current_user_id,
+        BindUser.status == 0,
+    ).first()
+    if not current_bind_user:
+        abort(403, "can not found bind user by id")
+
+    new_team_member = TeamMember(
+        id=ObjID.new_id(),
+        team_id=new_team.id,
+        code_user_id=current_bind_user.id,
+        im_user_id=None,
+    )
+
+    db.session.add(new_team_member)
+    db.session.commit()
+
+    return new_team
+
+
+def create_code_application(team_id: str, installation_id: str) -> CodeApplication:
+    """Create a code application.
+
+    Args:
+        team_id (str): Team ID.
+        installation_id (str): GitHub App installation ID.
+
+    Returns:
+        CodeApplication: CodeApplication object.
+    """
+
+    new_code_application = CodeApplication(
+        id=ObjID.new_id(),
+        team_id=team_id,
+        installation_id=installation_id,
+    )
+
+    db.session.add(new_code_application)
+    db.session.commit()
+
+    return new_code_application
+
+
 def save_im_application(
     team_id, platform, app_id, app_secret, encrypt_key, verification_token
 ):
@@ -153,4 +223,4 @@ def save_im_application(
                 ),
             )
         )
-        db.session.commit()
+        db.session.commit()

server/routes/github.py

@@ -1,11 +1,15 @@
-import logging
 import os
 
 from app import app
 from flask import Blueprint, abort, redirect, request, session
-from model.schema import Team
+from model.team import create_code_application, create_team
 from utils.auth import authenticated
-from utils.github.common import get_installation_token, get_jwt, verify_github_signature
+from utils.github.application import (
+    get_installation_info,
+    get_installation_token,
+    get_jwt,
+    verify_github_signature,
+)
 from utils.user import register
 
 bp = Blueprint("github", __name__, url_prefix="/api/github")
@@ -24,8 +28,37 @@ def github_install():
             f"https://github.com/apps/{os.environ.get('GITHUB_APP_NAME')}/installations/new"
         )
 
+    # 创建 Team
+    print(f"installation_id {installation_id}")
 
-@bp.route("/register", methods=["GET"])
+    jwt = get_jwt(
+        os.environ.get("GITHUB_APP_PRIVATE_KEY_PATH"),
+        os.environ.get("GITHUB_APP_ID"),
+    )
+    installation_token = get_installation_token(jwt, installation_id)
+    if installation_token is None:
+        app.logger.error("Failed to get installation token.")
+        return "Failed to get installation token."
+
+    app_info = get_installation_info(jwt, installation_id)
+    if app_info is None:
+        app.logger.error("Failed to get installation info.")
+        return "Failed to get installation info."
+
+    # 判断安装者的身份是用户还是组织
+    type = app_info["account"]["type"]
+    if type == "user":
+        app.logger.error("User is not allowed to install.")
+        # TODO: 定义与前端的交互数据格式
+        return "User is not allowed to install."
+
+    team = create_team(app_info)
+    code_application = create_code_application(team.id, installation_id)
+
+    return app_info
+
+
+@bp.route("/oauth", methods=["GET"])
 def github_register():
     """GitHub OAuth register.
 
@@ -57,9 +90,9 @@ def github_hook():
 
     x_github_event = request.headers.get("x-github-event", None)
 
-    logging.info(x_github_event)
+    app.logger.info(x_github_event)
 
-    logging.debug(request.json)
+    app.logger.debug(request.json)
 
     return "Receive Success!"
 

server/utils/github/account.py

@@ -0,0 +1,69 @@
+import httpx
+from app import app
+
+
+def get_user_info(access_token: str) -> dict | None:
+    """Get user info by access token.
+
+    Args:
+        access_token (str): The user access token.
+
+    Returns:
+        dict: User info.
+    """
+
+    with httpx.Client() as client:
+        response = client.get(
+            "https://api.github.com/user",
+            headers={
+                "Accept": "application/vnd.github.v3+json",
+                "Authorization": f"token {access_token}",
+            },
+        )
+        if response.status_code != 200:
+            app.logger.debug(f"Failed to get user info. {response.text}")
+            return None
+
+        user_info = response.json()
+        return user_info
+
+    app.logger.debug("Failed to get user info.")
+    return None
+
+
+def get_email(access_token: str) -> str | None:
+    """Get user email by access token.
+
+    Args:
+        access_token (str): The user access token.
+
+    Returns:
+        str: User email.
+    """
+
+    with httpx.Client() as client:
+        response = client.get(
+            "https://api.github.com/user/emails",
+            headers={
+                "Accept": "application/vnd.github.v3+json",
+                "Authorization": f"Bearer {access_token}",
+                "X-GitHub-Api-Version": "2022-11-28",
+            },
+        )
+        if response.status_code != 200:
+            app.logger.debug(f"Failed to get user email. {response.text}")
+            return None
+
+        user_emails = response.json()
+        if len(user_emails) == 0:
+            app.logger.debug("Failed to get user email.")
+            return None
+
+        for user_email in user_emails:
+            if user_email["primary"]:
+                return user_email["email"]
+
+        return user_emails[0]["email"]
+
+    app.logger.debug("Failed to get user email.")
+    return None